openapi schema

.env

 API_BASE=http://localhost
 API_PORT=8180
-API_PREFIX=/api
 EXPORT_DIR=exports/
 UPLOAD_DIR=uploads/
 CORS_ORIGIN=http://localhost:8080
\ No newline at end of file

app/config.py

+import logging
 import os
-from typing import Any
 
 import dotenv
-from fastapi import FastAPI
-from fastapi.routing import APIRoute
 
-ENV_CONFIG = {'API_PORT': '8180',
-              'API_PREFIX': None,
-              'CORS_ORIGIN': 'http://localhost:8080',
-              'EXPORT_DIR': 'exports/',
-              'UPLOAD_DIR': 'uploads/',
-              'root_path': ''
-              }
+logger = logging.getLogger(__name__)
+
+FASTAPI_CONFIG = {'API_PORT': '8180',
+                  'API_PREFIX': '',
+                  'CORS_ORIGIN': 'http://localhost:8080',
+                  }
+
+PATH_CONFIG = {
+    'EXPORT_DIR': 'exports/',
+    'UPLOAD_DIR': 'uploads/',
+}
+
+MITM_DB_CONFIG = {
+    'MITM_DATABASE_DIALECT': 'sqlite',
+    'MITM_DATABASE_USER': None,
+    'MITM_DATABASE_PASSWORD': None,
+    'MITM_DATABASE_HOST': None,
+    'MITM_DATABASE_PORT': None,
+    'MITM_DATABASE_DB': 'db/database.sqlite',
+}
+
+ENV_CONFIG = FASTAPI_CONFIG | PATH_CONFIG | MITM_DB_CONFIG
 
 
 def read_env() -> dict[str, str | None]:
     dotenv.load_dotenv(override=False)
-    return {k: os.environ.get(k, dv) for k, dv in ENV_CONFIG.items()}
+    return {k: os.getenv(k, dv) for k, dv in ENV_CONFIG.items()}
 
 
 app_cfg = read_env()
+
+logger.info(f'Loaded app config:\n{app_cfg}')

app/db/setup.py

-from contextlib import asynccontextmanager
+import logging
 
-import pydantic
-import sqlmodel
-from fastapi import FastAPI
-from mitm_tooling.definition import MITM
-from mitm_tooling.transformation.superset.definitions import SupersetAssetsDef
-from sqlalchemy.orm import Session
 import sqlalchemy as sa
+from mitm_tooling.utilities.python_utils import pick_from_mapping
 from sqlalchemy import create_engine
-from sqlmodel import SQLModel, Field
 
-engine = create_engine('sqlite:///db/database.db', execution_options=dict(check_same_thread=False))
+from ..config import app_cfg
 
+logger = logging.getLogger()
+MITM_DATABASE_URL = sa.engine.URL.create(*pick_from_mapping(app_cfg, ['MITM_DATABASE_DIALECT', 'MITM_DATABASE_USER',
+                                                                      'MITM_DATABASE_PASSWORD', 'MITM_DATABASE_HOST',
+                                                                      'MITM_DATABASE_PORT', 'MITM_DATABASE_DB'], flatten=True))
 
-class ImportedMitM(SQLModel):
-    id: int = Field(sa_column_kwargs={'autoincrement': True}, primary_key=True, nullable=False)
-    superset_id: int
-    uuid: str
-    mitm: MITM
-    asset_def: SupersetAssetsDef
-    superset_response: pydantic.JsonValue
+execution_options = {}
+if MITM_DATABASE_URL.get_dialect().name == 'sqlite':
+    execution_options.update(check_same_thread=False)
+
+engine = create_engine(MITM_DATABASE_URL, execution_options=execution_options)
+logger.info(f'Setting up MITM DB @ {MITM_DATABASE_URL}')
+
+#with engine.connect() as connection:
+#    connection.execute(sa.text('SELECT 1'))

app/main.py

-from contextlib import asynccontextmanager
-from typing import Annotated
-
-from fastapi import FastAPI, Depends
-from fastapi.routing import APIRoute
+from fastapi import FastAPI
 
+from .config import app_cfg
 from .dependencies.startup import lifecycle, use_route_names_as_operation_ids
 from .routers import upload, definitions
 
-from .config import app_cfg
+app = FastAPI(title='SupersetMitMService', lifespan=lifecycle, root_path=app_cfg['API_PREFIX'])
 
-app = FastAPI(title='SupersetMitMService', lifespan=lifecycle, root_path=app_cfg['root_path'])
 app.include_router(definitions.router)
 app.include_router(upload.router)
 

app/models/imported_mitm.py

+import sqlmodel
+from mitm_tooling.definition import MITM
+from mitm_tooling.transformation.superset.definitions import SupersetAssetsDef
+from sqlmodel import SQLModel
+import pydantic
+
+class ImportedMitM(SQLModel):
+    id: int = sqlmodel.Field(sa_column_kwargs={'autoincrement': True}, primary_key=True, nullable=False)
+    superset_id: int
+    uuid: str
+    mitm: MITM
+    asset_def: SupersetAssetsDef
+    superset_response: pydantic.JsonValue

pyproject.toml

@@ -11,16 +11,18 @@ python = "^3.12,<3.14"
 python-dotenv = "^1.0.1"
 pyyaml = "*"
 
+pydantic = "*"
 fastapi = { version = "^0.115.8", extras = ["standard"] }
 uvicorn = "^0.34.0"
 python-multipart = "*"
 sqlalchemy = "^2.0.38"
 sqlmodel = "^0.0.22"
 mitm-tooling = { version = "*" }
-
-#[tool.poetry.dev-dependencies]
 #mitm-tooling = { version = "*", source = "local", develop = true }
 
+[tool.poetry.dev.dependencies]
+datamodel-code-generator = "*"
+
 [[tool.poetry.source]]
 name = "local"
 url = "file:///C:/Users/leah/PycharmProjects/mitm-tooling"

schema/gen_open_api_schema.py

 import json
 
 import yaml
-from fastapi import FastAPI
 from fastapi.openapi.utils import get_openapi
-from fastapi.routing import APIRoute
 
 from app.main import app
 
-
-def use_route_names_as_operation_ids(app: FastAPI) -> None:
-    """
-    Simplify operation IDs so that generated API clients have simpler function
-    names.
-
-    Should be called only after all routes have been added.
-    """
-    for route in app.routes:
-        if isinstance(route, APIRoute):
-            route.operation_id = route.name
-
-
-use_route_names_as_operation_ids(app)
-
-with open('openapi.yaml', 'w') as f:
-    yaml.dump(get_openapi(
-        title=app.title,
-        version=app.version,
-        openapi_version=app.openapi_version,
-        description=app.description,
-        routes=app.routes
-    ), f)
-with open('openapi.json', 'w') as f:
-    json.dump(get_openapi(
-        title=app.title,
-        version=app.version,
-        openapi_version=app.openapi_version,
-        description=app.description,
-        routes=app.routes
-    ), f)
\ No newline at end of file
+if __name__ == "__main__":
+    with open('openapi.yaml', 'w') as f:
+        yaml.dump(get_openapi(
+            title=app.title,
+            version=app.version,
+            openapi_version=app.openapi_version,
+            description=app.description,
+            routes=app.routes
+        ), f)
+    with open('openapi.json', 'w') as f:
+        json.dump(get_openapi(
+            title=app.title,
+            version=app.version,
+            openapi_version=app.openapi_version,
+            description=app.description,
+            routes=app.routes
+        ), f)