Fix 143 --refactor upload to ckan, check for package existence before… (#144)

* Fix 143 --refactor upload to ckan, check for package existence before create

* Logging using stdout

Author: thepsalmist

docker-compose.yml

@@ -35,6 +35,9 @@ services:
       SENSORSAFRICA_FLOWER_ADMIN_USERNAME: ${SENSORSAFRICA_FLOWER_ADMIN_USERNAME:-admin}
       SENSORSAFRICA_FLOWER_ADMIN_PASSWORD: ${SENSORSAFRICA_FLOWER_ADMIN_PASSWORD:-password}
       DOKKU_APP_NAME: ${DOKKU_APP_NAME:-sensorsafrica}
+      CKAN_ARCHIVE_API_KEY: ${CKAN_ARCHIVE_API_KEY}
+      CKAN_ARCHIVE_OWNER_ID: ${CKAN_ARCHIVE_OWNER_ID}
+      CKAN_ARCHIVE_URL: ${CKAN_ARCHIVE_URL}
     depends_on:
       - postgres
       - rabbitmq

requirements.txt

@@ -6,7 +6,7 @@ timeago==1.0.10
 flower==0.9.5
 tornado<6
 sentry-sdk==1.14.0
-celery==4.2.1
+celery==4.3.0
 gevent==1.2.2
 greenlet==0.4.12
 whitenoise==4.1.2

sensorsafrica/management/commands/upload_to_ckan.py

@@ -45,17 +45,28 @@ def handle(self, *args, **options):
             if not timestamp or not timestamp['timestamp__min'] or not timestamp['timestamp__max']:
                 continue
 
+            package_name = f"sensorsafrica-airquality-archive-{slugify(city)}"
+            package_title = f"sensors.AFRICA Air Quality Archive {city}"
+
             try:
-                package = ckan.action.package_create(
-                    owner_org=CKAN_ARCHIVE_OWNER_ID,
-                    name="sensorsafrica-airquality-archive-" + slugify(city),
-                    title="sensors.AFRICA Air Quality Archive " + city,
-                    groups=[{"name": "sensorsafrica-airquality-archive"}]
-                )
-            except ckanapi.ValidationError:
-                package = ckan.action.package_show(
-                    id="sensorsafrica-airquality-archive-" + slugify(city)
-                )
+                package = ckan.action.package_show(id=package_name)
+                #To Do:xavier Implement Logging
+                self.stdout.write(f"Package '{package_name}' already exists. Skipping creation.")
+            except ckanapi.NotFound:
+                try:
+                    package = ckan.action.package_create(
+                        owner_org=CKAN_ARCHIVE_OWNER_ID,
+                        name=package_name,
+                        title=package_title,
+                        groups=[{"name": "sensorsafrica-airquality-archive"}]
+                    )
+                    self.stdout.write("Created new package '%s' for city." % city)
+                except ckanapi.ValidationError as e:
+                    self.stdout.write(f"Validation error creating package for city %s: %s" %city %e)
+                    continue
+            except Exception as e:
+                self.stdout.write(f"Unexpected error fetching package for city '{city}': {e}")
+                continue
 
             resources = package["resources"]
 
@@ -89,6 +100,7 @@ def handle(self, *args, **options):
                         timestamp__year=date.year,
                         sensordatavalues__value__isnull=False,
                     )
+                    .select_related("sensor","location")
                     .values(
                         "sensor__id",
                         "sensor__sensor_type__name",
@@ -107,16 +119,16 @@ def handle(self, *args, **options):
                         month=calendar.month_name[date.month], year=date.year
                     )
                     fp = tempfile.NamedTemporaryFile(mode="w+b", suffix=".csv")
-                    self._write_file(fp, qs)
-                    filepath = fp.name
-                   
-                    self._create_or_update_resource(
-                        resource_name, filepath, resources, ckan, package
-                    )
+                    try:
+                        self._write_file(fp, qs)
+                        filepath = fp.name
+                        self._create_or_update_resource(
+                            resource_name, filepath, resources, ckan, package
+                        )
+                    finally:
+                        # Cleanup temp file
+                        fp.close()
 
-                    # Cleanup temp file
-                    fp.close()
-                 
                     # Don't DDOS openAFRICA
                     time.sleep(5)
 
@@ -128,6 +140,8 @@ def handle(self, *args, **options):
                     tzinfo=pytz.UTC,
                 )
 
+        self.stdout.write("Data upload completed successfully.")
+
     @staticmethod
     def _write_file(fp, qs):
         fp.write(