FastAPI-ArchiPy-Boilerplate/manage.py at master · SyntaxArc/FastAPI-ArchiPy-Boilerplate · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
import logging
from contextlib import asynccontextmanager

import uvicorn
from archipy.adapters.sqlite.sqlalchemy.adapters import AsyncSQLiteSQLAlchemyAdapter

from archipy.helpers.utils.app_utils import AppUtils
from archipy.models.entities import BaseEntity
from fastapi import FastAPI

from src.configs.containers import ServiceContainer
from src.configs.dispatcher import set_dispatch_routes
from src.configs.runtime_config import RuntimeConfig


# TODO remove this method in production
# This is only for this boilerplate, don`t use in production environment
# Set up database schema with sync adapter
logging.info("Creating database schema with sync adapter")
@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup code
    logging.info("Creating database schema with async adapter")
    await async_schema_setup()
    yield
    # Shutdown code would go here if needed


container: ServiceContainer = ServiceContainer()
container.wire(packages=["src.controllers"])

app: FastAPI = AppUtils.create_fastapi_app(lifespan=lifespan)
app.container = container
set_dispatch_routes(app)

async def async_schema_setup():
    """Set up database schema for async adapter."""
    # Use AsyncEngine.begin() for proper transaction handling
    adapter = AsyncSQLiteSQLAlchemyAdapter()
    async with adapter.session_manager.engine.begin() as conn:
        # Drop all tables (but only if they exist)
        await conn.run_sync(BaseEntity.metadata.drop_all)
        # Create all tables
        await conn.run_sync(BaseEntity.metadata.create_all)


if __name__ == "__main__":
    logging.basicConfig(
        level=RuntimeConfig.global_config().ENVIRONMENT.log_level,
        filename="../siteLogs.log",
        format="{'time':'%(asctime)s', 'name': '%(name)s', \
        'level': '%(levelname)s', 'message': '%(message)s'}",
    )

    uvicorn.run(
        app="manage:app",
        access_log=RuntimeConfig.global_config().FASTAPI.ACCESS_LOG,
        backlog=RuntimeConfig.global_config().FASTAPI.BACKLOG,
        date_header=RuntimeConfig.global_config().FASTAPI.DATE_HEADER,
        forwarded_allow_ips=RuntimeConfig.global_config().FASTAPI.FORWARDED_ALLOW_IPS,
        host=RuntimeConfig.global_config().FASTAPI.SERVE_HOST,
        limit_concurrency=RuntimeConfig.global_config().FASTAPI.LIMIT_CONCURRENCY,
        limit_max_requests=RuntimeConfig.global_config().FASTAPI.LIMIT_MAX_REQUESTS,
        port=RuntimeConfig.global_config().FASTAPI.SERVE_PORT,
        proxy_headers=RuntimeConfig.global_config().FASTAPI.PROXY_HEADERS,
        reload=RuntimeConfig.global_config().FASTAPI.RELOAD,
        server_header=RuntimeConfig.global_config().FASTAPI.SERVER_HEADER,
        timeout_graceful_shutdown=RuntimeConfig.global_config().FASTAPI.TIMEOUT_GRACEFUL_SHUTDOWN,
        timeout_keep_alive=RuntimeConfig.global_config().FASTAPI.TIMEOUT_KEEP_ALIVE,
        workers=RuntimeConfig.global_config().FASTAPI.WORKERS_COUNT,
        ws_max_size=RuntimeConfig.global_config().FASTAPI.WS_MAX_SIZE,
        ws_per_message_deflate=RuntimeConfig.global_config().FASTAPI.WS_PER_MESSAGE_DEFLATE,
        ws_ping_interval=RuntimeConfig.global_config().FASTAPI.WS_PING_INTERVAL,
        ws_ping_timeout=RuntimeConfig.global_config().FASTAPI.WS_PING_TIMEOUT,
    )