Fix task initialization

.gitignore

@@ -90,3 +90,7 @@ downloads.db
 *.gz
 
 data/*
+
+celerybeat-schedule.bak
+celerybeat-schedule.dat
+celerybeat-schedule.dir

conda/dev.yaml

@@ -3,8 +3,8 @@ channels:
   - conda-forge
   - defaults
 dependencies:
-  - docker-compose #
-  - celery #
+  - docker-compose
+  - celery
   - make
   - poetry
   - python=3.9

docker/celery/run.sh

@@ -1,3 +1,3 @@
 #!/bin/bash
 
-exec celery -A satellite_weather_downloader.celery_app.celeryapp beat -l DEBUG
+exec celery --workdir satellite_weather_downloader/celery_app --config celeryconfig -A tasks worker -B --loglevel=DEBUG

pyproject.toml

@@ -35,6 +35,7 @@ MetPy = "^1.3.1"
 SQLAlchemy = "^1.4.41"
 python-dotenv = "^0.21.0"
 psycopg2-binary = "^2.9.4"
+netCDF4 = "^1.6.1"
 
 [tool.poetry.dev-dependencies]
 pytest = "^5.2"

satellite_weather_downloader/celery_app/__init__.py

@@ -0,0 +1,5 @@
+# This will make sure the app is always imported when
+# Django starts so that shared_task will use this app.
+from .celeryapp import app as celery_app
+
+__all__ = ('celery_app',)

satellite_weather_downloader/celery_app/celeryapp.py

@@ -5,6 +5,16 @@
 app = Celery("satellite_weather_downloader")
 
 app.config_from_object("satellite_weather_downloader.celery_app.celeryconfig")
+app.autodiscover_tasks()
+
+
+@app.task(bind=True)
+def debug_task(self):
+    print(f'Request: {self.request!r}')
+
 
 if __name__ == "__main__":
-    app.start()
+    worker = app.Worker(
+        include=['satellite_weather_downloader.celery_app.tasks']
+    )
+    worker.start()

satellite_weather_downloader/celery_app/tasks.py

@@ -3,6 +3,7 @@
 import logging
 import os
 
+
 from sqlalchemy import create_engine
 from celery.schedules import crontab
 from datetime import datetime, timedelta
@@ -18,8 +19,8 @@
 
 PSQL_USER = os.getenv("POSTGRES_USER")
 PSQL_PASSWORD = os.getenv("POSTGRES_PASSWORD")
-HOST = os.getenv("PSQL_HOST")
-PORT = os.getenv("PSQL_PORT")
+HOST = os.getenv("POSTGRES_HOST")
+PORT = os.getenv("POSTGRES_PORT")
 DBASE = os.getenv("POSTGRES_DATABASE")
 
 
@@ -95,6 +96,7 @@ def reanalysis_fetch_data_daily():
     for geocode in geocodes:
         row = netcdf_to_dataframe(data, geocode)
         cope_df = cope_df.merge(row, on=list(cope_df.columns), how='outer')
+        logging.info(f'{geocode} added to dataframe')
 
     cope_df = cope_df.set_index('date')