Towards django RQ

app_urls/api/admin.py

@@ -0,0 +1,3 @@
+from django.contrib import admin
+
+# Register your models here.

app_urls/api/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class ApiConfig(AppConfig):
+    default_auto_field = 'django.db.models.BigAutoField'
+    name = 'api'

app_urls/api/migrations/0001_initial.py

@@ -0,0 +1,132 @@
+# Generated by Django 5.1.7 on 2025-03-07 16:56
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='Feed',
+            fields=[
+                ('id', models.SmallAutoField(primary_key=True, serialize=False)),
+                ('rss_feed', models.TextField(unique=True)),
+            ],
+            options={
+                'db_table': 'feed',
+                'managed': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='Search',
+            fields=[
+                ('id', models.SmallAutoField(primary_key=True, serialize=False)),
+                ('keyword_search', models.TextField(unique=True)),
+            ],
+            options={
+                'db_table': 'search',
+                'managed': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='Source',
+            fields=[
+                ('id', models.SmallAutoField(primary_key=True, serialize=False)),
+                ('source', models.TextField(unique=True)),
+            ],
+            options={
+                'db_table': 'source',
+                'managed': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='StatusPatternMatching',
+            fields=[
+                ('pattern', models.TextField(primary_key=True, serialize=False)),
+                ('priority', models.SmallIntegerField()),
+                ('status', models.TextField()),
+            ],
+            options={
+                'db_table': 'status_pattern_matching',
+                'managed': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='Urls',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('url', models.TextField(unique=True)),
+                ('ts_fetch', models.DateTimeField()),
+                ('status', models.TextField()),
+            ],
+            options={
+                'db_table': 'urls',
+                'managed': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='WebsiteOfInterest',
+            fields=[
+                ('id', models.SmallAutoField(primary_key=True, serialize=False)),
+                ('url_host', models.TextField(unique=True)),
+            ],
+            options={
+                'db_table': 'website_of_interest',
+                'managed': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='WebsiteToFilter',
+            fields=[
+                ('id', models.SmallAutoField(primary_key=True, serialize=False)),
+                ('url_host', models.TextField(unique=True)),
+            ],
+            options={
+                'db_table': 'website_to_filter',
+                'managed': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='UrlContent',
+            fields=[
+                ('id_url', models.OneToOneField(db_column='id_url', on_delete=django.db.models.deletion.DO_NOTHING, primary_key=True, serialize=False, to='api.urls')),
+                ('date_published', models.DateTimeField(blank=True, null=True)),
+                ('title', models.TextField(blank=True, null=True)),
+                ('description', models.TextField(blank=True, null=True)),
+                ('content', models.TextField(blank=True, null=True)),
+                ('tags', models.TextField(blank=True, null=True)),
+                ('authors', models.TextField(blank=True, null=True)),
+                ('image_urls', models.TextField(blank=True, null=True)),
+            ],
+            options={
+                'db_table': 'url_content',
+                'managed': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='UrlsDuplicate',
+            fields=[
+                ('id_url_canonical', models.OneToOneField(db_column='id_url_canonical', on_delete=django.db.models.deletion.DO_NOTHING, primary_key=True, serialize=False, to='api.urls')),
+            ],
+            options={
+                'db_table': 'urls_duplicate',
+                'managed': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='UrlsSource',
+            fields=[
+                ('id_url', models.OneToOneField(db_column='id_url', on_delete=django.db.models.deletion.DO_NOTHING, primary_key=True, serialize=False, to='api.urls')),
+            ],
+            options={
+                'db_table': 'urls_source',
+                'managed': False,
+            },
+        ),
+    ]

app_urls/api/models.py

@@ -0,0 +1,101 @@
+from django.db import models
+
+# Create your models here.
+class Feed(models.Model):
+    id = models.SmallAutoField(primary_key=True)
+    rss_feed = models.TextField(unique=True)
+
+    class Meta:
+        managed = False
+        db_table = 'feed'
+
+
+class Search(models.Model):
+    id = models.SmallAutoField(primary_key=True)
+    keyword_search = models.TextField(unique=True)
+
+    class Meta:
+        managed = False
+        db_table = 'search'
+
+
+class Source(models.Model):
+    id = models.SmallAutoField(primary_key=True)
+    source = models.TextField(unique=True)
+
+    class Meta:
+        managed = False
+        db_table = 'source'
+
+
+class StatusPatternMatching(models.Model):
+    pattern = models.TextField(primary_key=True)
+    priority = models.SmallIntegerField()
+    status = models.TextField()  # This field type is a guess.
+
+    class Meta:
+        managed = False
+        db_table = 'status_pattern_matching'
+
+
+class UrlContent(models.Model):
+    id_url = models.OneToOneField('Urls', models.DO_NOTHING, db_column='id_url', primary_key=True)
+    date_published = models.DateTimeField(blank=True, null=True)
+    title = models.TextField(blank=True, null=True)
+    description = models.TextField(blank=True, null=True)
+    content = models.TextField(blank=True, null=True)
+    tags = models.TextField(blank=True, null=True)  # This field type is a guess.
+    authors = models.TextField(blank=True, null=True)  # This field type is a guess.
+    image_urls = models.TextField(blank=True, null=True)  # This field type is a guess.
+
+    class Meta:
+        managed = False
+        db_table = 'url_content'
+
+
+class Urls(models.Model):
+    url = models.TextField(unique=True)
+    ts_fetch = models.DateTimeField()
+    status = models.TextField()  # This field type is a guess.
+
+    class Meta:
+        managed = False
+        db_table = 'urls'
+
+
+class UrlsDuplicate(models.Model):
+    id_url_canonical = models.OneToOneField(Urls, models.DO_NOTHING, db_column='id_url_canonical', primary_key=True)  # The composite primary key (id_url_canonical, id_url_duplicated) found, that is not supported. The first column is selected.
+    id_url_duplicated = models.ForeignKey(Urls, models.DO_NOTHING, db_column='id_url_duplicated', related_name='urlsduplicate_id_url_duplicated_set')
+
+    class Meta:
+        managed = False
+        db_table = 'urls_duplicate'
+        unique_together = (('id_url_canonical', 'id_url_duplicated'),)
+
+
+class UrlsSource(models.Model):
+    id_url = models.OneToOneField(Urls, models.DO_NOTHING, db_column='id_url', primary_key=True)  # The composite primary key (id_url, id_source) found, that is not supported. The first column is selected.
+    id_source = models.ForeignKey(Source, models.DO_NOTHING, db_column='id_source')
+
+    class Meta:
+        managed = False
+        db_table = 'urls_source'
+        unique_together = (('id_url', 'id_source'),)
+
+
+class WebsiteOfInterest(models.Model):
+    id = models.SmallAutoField(primary_key=True)
+    url_host = models.TextField(unique=True)
+
+    class Meta:
+        managed = False
+        db_table = 'website_of_interest'
+
+
+class WebsiteToFilter(models.Model):
+    id = models.SmallAutoField(primary_key=True)
+    url_host = models.TextField(unique=True)
+
+    class Meta:
+        managed = False
+        db_table = 'website_to_filter'

app_urls/api/tasks.py

@@ -0,0 +1,13 @@
+from django_rq import job
+import time
+import logging
+logger = logging.getLogger(__name__)
+
+@job
+def task_1(message):
+    logger.info("Message: {}".format(message))
+    try:
+        time.sleep(5)  # Simulate a long-running task
+        print(f"Task completed: {message}")
+    except Exception as e:
+        logger.error(e)

app_urls/api/tests.py

@@ -0,0 +1,3 @@
+from django.test import TestCase
+
+# Create your tests here.

app_urls/api/urls.py

@@ -0,0 +1,6 @@
+from django.urls import path
+from .views import trigger_task
+
+urlpatterns = [
+    path('trigger_task/', trigger_task, name='trigger_task')
+]

app_urls/api/views.py

@@ -0,0 +1,10 @@
+import django_rq
+from django.http import JsonResponse
+from .tasks import task_1
+
+def trigger_task(request):
+    """View that enqueues a task."""
+    queue = django_rq.get_queue('default')  # Get the default queue
+    job = queue.enqueue(task_1, "Hello from Django RQ!")
+
+    return JsonResponse({"message": "Task has been enqueued!", "job_id": job.id})