celery et context app task

.circleci/config.yml

@@ -7,11 +7,13 @@ jobs: # a collection of steps
       - image: circleci/python:3.6.6 # ...with this image as the primary container; this is where all `steps` will run        environment:
         environment:
           POSTGRES_URL: postgresql://${DB_NAME}_user:${DB_NAME}_password@localhost:5432/${DB_NAME}
+          REDIS_URL: redis://localhost:6379/0
       - image: circleci/postgres:9.6.2
         environment:
           POSTGRES_DB: ${DB_NAME}
           POSTGRES_USER: ${DB_NAME}_user
           POSTGRES_PASSWORD: ${DB_NAME}_password
+      - image: circleci/redis:6.0.9
 
     steps:
       - checkout # special step to check out source code to the working directory

Dockerfile

@@ -1,9 +1,9 @@
 FROM python:3.7
 
 ENV PYTHONUNBUFFERED 1
-WORKDIR /opt/apiweb
+WORKDIR /opt/api
 
 COPY ./requirements.txt ./
 COPY ./requirements-test.txt ./
-RUN pip install --upgrade pip==20.3
+RUN pip install --upgrade pip==20.3.3
 RUN pip install -r ./requirements.txt -r requirements-test.txt

api/checkhealth.py

@@ -0,0 +1,67 @@
+# pylint: disable=W0611
+
+import os
+from time import sleep
+from flask import Flask
+from random import random
+from sqlalchemy_api_handler import ApiHandler
+from sqlalchemy_api_handler.utils import logger
+from sqlalchemy.exc import OperationalError
+
+from api.repository.checks import check_from_model
+from api.utils.config import VERSION
+from api.utils.setup import setup
+from api.utils.database import db
+
+
+SLEEP_TIME = 1
+
+FLASK_APP = Flask(__name__)
+
+
+setup(FLASK_APP,
+      with_cors=False)
+
+
+TABLE_NAME = f'check_health_{int(1000*random())}'
+class CheckHealth(db.Model, ApiHandler):
+    __tablename__ = TABLE_NAME
+    pass
+
+
+def check_from_model(model):
+    database_working = False
+    version = 'Api {} : '.format(VERSION)
+    try:
+        model.query.limit(1).all()
+        database_working = True
+        state = 'database is ok.'
+    except Exception as e:
+        logger.critical(str(e))
+        state = 'database is not ok.'
+    output = '{}{}'.format(version, state)
+
+    return database_working, output
+
+IS_DATABASE_CONNECT_OK = False
+while not IS_DATABASE_CONNECT_OK:
+    try:
+        CheckHealth.__table__.drop(db.session.bind, checkfirst=True)
+        CheckHealth.__table__.create(db.session.bind)
+        db.session.commit()
+    except OperationalError:
+        print('Could not connect to postgres db... Retry in {}s...'.format(SLEEP_TIME))
+        sleep(SLEEP_TIME)
+        continue
+    print('Connection to postgres db is okay.')
+    IS_DATABASE_CONNECT_OK = True
+
+IS_DATABASE_HEALTH_OK = False
+while not IS_DATABASE_HEALTH_OK:
+    IS_DATABASE_HEALTH_OK = check_from_model(CheckHealth)[0]
+    db.session.execute(f'DROP TABLE {TABLE_NAME};')
+    db.session.commit()
+    if not IS_DATABASE_HEALTH_OK:
+        print('Could not check database health... Retry in {}s...'.format(SLEEP_TIME))
+    else:
+        print('Database health is ok.')

api/models/__init__.py

@@ -13,6 +13,7 @@ def import_models():
     from api.models.scope import Scope
     from api.models.stock import Stock
     from api.models.tag import Tag
+    from api.models.task import Task
     from api.models.time_interval import TimeInterval
     from api.models.user import User
     from api.models.user_offerer import UserOfferer

api/models/task.py

@@ -0,0 +1,10 @@
+from sqlalchemy_api_handler import ApiHandler
+from sqlalchemy_api_handler.mixins import TaskMixin
+
+from api.utils.database import db
+
+
+class Task(ApiHandler,
+            db.Model,
+            TaskMixin):
+    pass

api/tasks/__init__.py

@@ -0,0 +1,2 @@
+def import_tasks():
+    import api.tasks.hello

api/tasks/hello.py

@@ -0,0 +1,9 @@
+from sqlalchemy_api_handler import ApiHandler
+
+from api.utils.celery import CELERY_APP
+
+
+@CELERY_APP.task
+def print_text(text):
+    print(f'Hello {text}')
+    return { 'text': text }

api/utils/celery.py

@@ -0,0 +1,17 @@
+import os
+import celery
+
+from api.utils.config import APP_NAME
+
+
+CELERY_TASKS_LIMIT = int(os.environ.get('CELERY_TASKS_LIMIT') or 10)
+LOCALHOST_REDIS_URL = f'redis://apiredisdb:6379/0'
+REDIS_URL = os.environ.get('REDIS_URL', LOCALHOST_REDIS_URL)
+
+
+CELERY_APP = celery.Celery('{}-tasks'.format(APP_NAME),
+                           backend=REDIS_URL,
+                           broker=REDIS_URL,
+                           broker_pool_limit=CELERY_TASKS_LIMIT)
+CELERY_APP.conf.task_default_queue = 'default'
+CELERY_APP.is_disabled = False

api/utils/config.py

@@ -1,6 +1,15 @@
 import os
+import sys
 
 APP_NAME = os.environ.get('APP_NAME', '')
 COMMAND_NAME = os.environ.get('COMMAND_NAME', '')
 PORT = int(os.environ.get('PORT', 5000))
 API_URL = 'http://api:{}'.format(PORT)
+
+IS_SCHEDULER = sys.argv[0].endswith('celery') \
+               and len(sys.argv) > 3 and \
+               sys.argv[2] == 'beat'
+IS_WORKER = sys.argv[0].endswith('celery') \
+            and len(sys.argv) > 3 \
+            and sys.argv[2] == 'worker'
+IS_APP = not IS_SCHEDULER and not IS_WORKER

api/utils/database.py

@@ -1,11 +1,17 @@
+import os
 from sqlalchemy import orm
 from flask_sqlalchemy import SQLAlchemy
-
 from sqlalchemy_api_handler import ApiHandler
 from sqlalchemy_api_handler.utils import logger
 
+from api.utils.config import APP_NAME
+
+
+LOCALHOST_POSTGRES_URL = f'postgresql://{APP_NAME}_user:{APP_NAME}_password@apipostgresdb/{APP_NAME}_apipostgres'
+POSTGRES_URL = os.environ.get('POSTGRES_URL', LOCALHOST_POSTGRES_URL)
 
 db = SQLAlchemy()
+ApiHandler.set_db(db)
 
 
 def delete():

api/utils/setup.py

@@ -2,30 +2,44 @@
 from flask_login import LoginManager
 from sqlalchemy_api_handler import ApiHandler
 
-
 from api.models import import_models
-from api.utils.database import db
+from api.utils.config import IS_APP, \
+                             IS_WORKER
+from api.utils.database import POSTGRES_URL, db
 
 
-def setup(flask_app):
+def setup(flask_app,
+          with_login_manager=False):
     flask_app.config['SECRET_KEY'] = '@##&6cweafhv3426445'
-    flask_app.config['SQLALCHEMY_DATABASE_URI'] = os.environ.get('POSTGRES_URL')
+    flask_app.config['SQLALCHEMY_DATABASE_URI'] = POSTGRES_URL
     flask_app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
     flask_app.config['TESTING'] = True
 
     db.init_app(flask_app)
-    ApiHandler.set_db(db)
+
+    @flask_app.teardown_request
+    def remove_db_session(exc):
+        try:
+            db.session.remove()
+        except AttributeError:
+            pass
 
     flask_app.app_context().push()
     import_models()
 
-    flask_app.config['SESSION_COOKIE_HTTPONLY'] = not flask_app.config['TESTING']
-    flask_app.config['SESSION_COOKIE_SECURE'] = False
-    flask_app.config['REMEMBER_COOKIE_HTTPONLY'] = not flask_app.config['TESTING']
-    if not flask_app.config['TESTING']:
-        flask_app.config['PERMANENT_SESSION_LIFETIME'] = 90 * 24 * 3600
-        flask_app.config['REMEMBER_COOKIE_DURATION'] = 90 * 24 * 3600
-        flask_app.config['REMEMBER_COOKIE_SECURE'] = True
-    login_manager = LoginManager()
-    login_manager.init_app(flask_app)
-    import api.utils.login_manager
+    if with_login_manager:
+        from flask_login import LoginManager
+        flask_app.config['SESSION_COOKIE_HTTPONLY'] = not flask_app.config['TESTING']
+        flask_app.config['SESSION_COOKIE_SECURE'] = False
+        flask_app.config['REMEMBER_COOKIE_HTTPONLY'] = not flask_app.config['TESTING']
+        if not flask_app.config['TESTING']:
+            flask_app.config['PERMANENT_SESSION_LIFETIME'] = 90 * 24 * 3600
+            flask_app.config['REMEMBER_COOKIE_DURATION'] = 90 * 24 * 3600
+            flask_app.config['REMEMBER_COOKIE_SECURE'] = True
+        login_manager = LoginManager()
+        login_manager.init_app(flask_app)
+        import api.utils.login_manager
+
+    if IS_APP or IS_WORKER:
+        from api.utils.celery import CELERY_APP
+        ApiHandler.set_celery(CELERY_APP, flask_app)

commands/dev.py

@@ -13,7 +13,7 @@ class DevCommand(Command):
     capture_all_args = True
 
     def run(self, args):
-        file_path = '{}{}'.format('/opt/apiweb/', args[0])
+        file_path = '{}{}'.format('/opt/api/', args[0])
         kwargs = {}
         if len(args) > 1:
             kwargs['argv'] = args[1:]