fix: test

Author: SKairinos

Pipfile

@@ -29,7 +29,7 @@ boto3 = "==1.34.162"
 
 [dev-packages]
 codeforlife = {version = "==0.22.12", extras = ["dev"]}
-boto3-stubs = {version = "==1.35.71", extras = ["ec2", "s3"]}
+boto3-stubs = {version = "==1.35.71", extras = ["essential"]}
 # codeforlife = {file = "../codeforlife-package-python", editable = true, extras = ["dev"]}
 # 🚫 Don't add [dev-packages] below that are inherited from the CFL package.
 

Pipfile.lock

@@ -7,12 +7,55 @@
 
 import os
 
-from codeforlife.app import StandaloneApplication
+# from codeforlife.app import StandaloneApplication
 from django.core.asgi import get_asgi_application
 from django.core.wsgi import get_wsgi_application
 
 os.environ.setdefault("DJANGO_SETTINGS_MODULE", "settings")
 
+# TODO: delete this
+# pylint: disable-all
+
+import multiprocessing
+import typing as t
+
+from django.core.management import call_command
+from gunicorn.app.base import BaseApplication  # type: ignore[import-untyped]
+
+
+# pylint: disable-next=abstract-method
+class StandaloneApplication(BaseApplication):
+    """A server for an app in a live environment.
+
+    Based off of:
+    https://gist.github.com/Kludex/c98ed6b06f5c0f89fd78dd75ef58b424
+    https://docs.gunicorn.org/en/stable/custom.html
+    """
+
+    def __init__(self, app: t.Callable):
+        call_command("migrate", interactive=False)
+
+        self.options = {
+            "bind": "0.0.0.0:8080",
+            # https://docs.gunicorn.org/en/stable/design.html#how-many-workers
+            "workers": 1,
+            "worker_class": "uvicorn.workers.UvicornWorker",
+        }
+        self.application = app
+        super().__init__()
+
+    def load_config(self):
+        config = {
+            key: value
+            for key, value in self.options.items()
+            if key in self.cfg.settings and value is not None
+        }
+        for key, value in config.items():
+            self.cfg.set(key.lower(), value)
+
+    def load(self):
+        return self.application
+
 
 if __name__ == "__main__":
     StandaloneApplication(app=get_asgi_application()).run()

application.py

@@ -67,97 +67,69 @@
 
 import json
 import logging
+import typing as t
 
 import boto3
+from codeforlife.types import JsonDict
 
+if t.TYPE_CHECKING:
+    from mypy_boto3_s3.client import S3Client
 
-def check_for_pointer_file(S3_APP_BUCKET, S3_APP_KEY):
-    s3 = boto3.client("s3")
 
-    pointer = s3.list_objects(
-        Bucket=S3_APP_BUCKET, Prefix=f"{S3_APP_KEY}/dbMetadata/pointer"
-    )
-
-    if pointer.get("Contents"):
-        resp = s3.get_object(
-            Bucket=S3_APP_BUCKET, Key=pointer["Contents"][0]["Key"]
-        )
-        return resp["Body"].read().decode("utf-8")
-
-    return None
-
-
-def construct_db_config(S3_APP_BUCKET, S3_KEY, DB_DATA):
-    s3 = boto3.client("s3")
-
-    print(f'connecting with bucket: "{S3_APP_BUCKET}", key: "{S3_KEY}"')
-    logging.info(f'connecting with bucket: "{S3_APP_BUCKET}", key: "{S3_KEY}"')
+AWS_S3_APP_BUCKET = os.getenv("aws_s3_app_bucket")
+AWS_S3_APP_FOLDER = os.getenv("aws_s3_app_folder")
+# AWS_REGION = os.getenv("aws_region")
+APP_ID = os.getenv("APP_ID")  # type: ignore[assignment]
+APP_VERSION = os.getenv("APP_VERSION")  # type: ignore[assignment]
 
-    cfg = s3.get_object(Bucket=S3_APP_BUCKET, Key=S3_KEY)
 
-    config = json.loads(cfg["Body"].read().decode("utf-8"))
+def get_databases():
+    if AWS_S3_APP_BUCKET and AWS_S3_APP_FOLDER and APP_ID:
+        key = f"{AWS_S3_APP_FOLDER}/dbMetadata/{APP_ID}/app.dbdata"
 
-    if config and config["DBEngine"] == "postgres":
-        DB_DATA["default"].update(
-            {
-                "NAME": config["Database"],
-                "USER": config["user"],
-                "PASSWORD": config["password"],
-                "HOST": config["Endpoint"],
-                "PORT": config["Port"],
-            }
+        print(
+            f'(print) connecting with bucket: "{AWS_S3_APP_BUCKET}", key: "{key}"'
+        )
+        logging.info(
+            f'(log) connecting with bucket: "{AWS_S3_APP_BUCKET}", key: "{key}"'
         )
 
-        return DB_DATA
-
+        s3: "S3Client" = boto3.client("s3")
+        db_data_object = s3.get_object(Bucket=AWS_S3_APP_BUCKET, Key=key)
 
-def load_db_config(S3_APP_BUCKET, S3_APP_KEY):
-    s3 = boto3.client("s3")
+        db_data: JsonDict = json.loads(
+            db_data_object["Body"].read().decode("utf-8")
+        )
 
-    default_dict = {
+        if not db_data or db_data["DBEngine"] != "postgres":
+            raise ConnectionAbortedError("Invalid database data.")
+
+        name = db_data["Database"]
+        user = db_data["user"]
+        password = db_data["password"]
+        host = db_data["Endpoint"]
+        port = db_data["Port"]
+    else:
+        name = os.getenv("DB_NAME", SERVICE_NAME)
+        user = os.getenv("DB_USER", "root")
+        password = os.getenv("DB_PASSWORD", "password")
+        host = os.getenv("DB_HOST", "localhost")
+        port = int(os.getenv("DB_PORT", "5432"))
+
+    return {
         "default": {
-            # "ENGINE": "django.db.backends.postgresql_psycopg2",
             "ENGINE": "django.db.backends.postgresql",
-            "NAME": "",
-            "USER": "",
-            "PASSWORD": "",
-            "HOST": "",
-            "PORT": "",
-            "OPTIONS": {
-                "connect_timeout": 300,
-            },
+            "NAME": name,
+            "USER": user,
+            "PASSWORD": password,
+            "HOST": host,
+            "PORT": port,
+            # "OPTIONS": {
+            #     "connect_timeout": 300,
+            # },
             "ATOMIC_REQUESTS": True,
         }
     }
 
-    link = check_for_pointer_file(S3_APP_BUCKET, S3_APP_KEY)
-    if link:
-        return construct_db_config(S3_APP_BUCKET, link, default_dict)
-
-    objs = s3.list_objects(
-        Bucket=S3_APP_BUCKET, Prefix=f"{S3_APP_KEY}/dbMetadata/"
-    )
-    for config_file in objs.get("Contents", []):
-        return construct_db_config(
-            S3_APP_BUCKET, config_file["Key"], default_dict
-        )
-
-
-S3_BUCKET = os.getenv("aws_s3_app_bucket")
-S3_PREFIX = os.getenv("aws_s3_app_folder")
-# AWS_REGION = os.getenv("aws_region")
 
-if S3_BUCKET and S3_PREFIX:
-    DATABASES = load_db_config(S3_BUCKET, S3_PREFIX)
-
-# DATABASES = {
-#     "default": {
-#         "ENGINE": "django.db.backends.postgresql",
-#         "NAME": os.getenv("DB_NAME", SERVICE_NAME),
-#         "HOST": os.getenv("DB_HOST", "localhost"),
-#         "PORT": int(os.getenv("DB_PORT", "5432")),
-#         # "USER": os.getenv("DB_USER", "root"),
-#         # "PASSWORD": os.getenv("DB_PASSWORD", "password"),
-#         "ATOMIC_REQUESTS": True,
-#     }
-# }
+DATABASES = get_databases()