20615450373

Committed 31 Dec 2025 08:41AM UTC coverage: 93.775%. First build

Build # 20615450373

Build Type

Pull #153

github

Committed by

web-flow

Commit Message

Merge 161984096 into b93cf7d42

Pull Request Pull Request #153: Modernization

Run Details

11 of 12 new or added lines in 3 files covered. (91.67%)

2772 of 2956 relevant lines covered (93.78%)

0.94 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

62.96

/api/src/app/core/setup.py

from contextlib import asynccontextmanager
from typing import Any, AsyncContextManager, AsyncGenerator, Callable

import anyio
from arq import create_pool
from arq.connections import RedisSettings
from fastapi import APIRouter, FastAPI
from fastapi.middleware.cors import CORSMiddleware

from ..middlewares.yaml_middleware import add_yaml_middleware_to_router
from .config import AppSettings, DatabaseSettings, RedisQueueSettings, settings
from .db.database import Base
from .db.database import async_engine as engine
from .utils import queue


# Function to create database tables
async def create_tables() -> None:
    """Create SQL tables."""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all, checkfirst=True)


async def create_redis_queue_pool() -> None:
    """Create Redis queue pool."""
    queue.pool = await create_pool(
        RedisSettings(
            host=settings.REDIS_QUEUE_HOST,
            port=settings.REDIS_QUEUE_PORT,
            password=settings.REDIS_QUEUE_PASSWORD,
        )
    )


async def close_redis_queue_pool() -> None:
    """Close Redis queue pool."""
    if queue.pool:
        await queue.pool.aclose()


async def set_threadpool_tokens(number_of_tokens: int = 100) -> None:
    """Set thread limit."""
    limiter = anyio.to_thread.current_default_thread_limiter()  # type: ignore[attr-defined]
    limiter.total_tokens = number_of_tokens


# Lifespan factory to manage app lifecycle events
def lifespan_factory(
    settings: DatabaseSettings | AppSettings,
    create_tables_on_start: bool = True,
) -> Callable[[FastAPI], AsyncContextManager[Any]]:
    """Create a lifespan async context manager for a FastAPI app."""

    @asynccontextmanager
    async def lifespan(app: FastAPI) -> AsyncGenerator[Any, None]:
        from asyncio import Event  # noqa: PLC0415

        initialization_complete = Event()
        app.state.initialization_complete = initialization_complete

        await set_threadpool_tokens()

        try:
            if isinstance(settings, DatabaseSettings) and create_tables_on_start:
                await create_tables()

            if isinstance(settings, RedisQueueSettings):
                await create_redis_queue_pool()

            initialization_complete.set()

            yield

        finally:
            if isinstance(settings, RedisQueueSettings):
                await close_redis_queue_pool()

    return lifespan


# Function to create the FastAPI app
def create_application(
    router: APIRouter,
    settings: DatabaseSettings | AppSettings,
    create_tables_on_start: bool = True,
    **kwargs: Any,  # noqa: ANN401
) -> FastAPI:
    """Create and configure a FastAPI application based on the provided settings.

    This function initializes a FastAPI application and configures it with various settings
    and handlers based on the type of the `settings` object provided.

    Args:
    ----
    router (APIRouter): The APIRouter object containing the routes to be included in the FastAPI application.

    settings: An instance representing the settings for configuring the FastAPI application.
        It determines the configuration applied:

        - AppSettings: Configures basic app metadata like name, description, contact, and license info.
        - DatabaseSettings: Adds event handlers for initializing database tables during startup.

    create_tables_on_start (bool): A flag to indicate whether to create database tables on application startup. Defaults to True.

    **kwargs (Any): Additional keyword arguments passed directly to the FastAPI constructor.

    Returns:
    -------
        FastAPI: A fully configured FastAPI application instance.

    The function configures the FastAPI application with different features and behaviors
    based on the provided settings. It includes setting up database connections, Redis pools
    for caching, queue, and rate limiting, client-side caching, and customizing the API documentation
    based on the environment settings.

    """
    # --- before creating application ---
    if isinstance(settings, AppSettings):
        to_update = {
            "title": settings.APP_NAME,
            "version": settings.APP_VERSION,
            "description": settings.APP_DESCRIPTION,
            "contact": {"name": settings.CONTACT_NAME, "email": settings.CONTACT_EMAIL},
            "license_info": {
                "name": settings.LICENSE_NAME,
                "url": settings.LICENSE_URL,
            },
        }
        kwargs.update(to_update)

    lifespan = lifespan_factory(settings, create_tables_on_start=create_tables_on_start)

    app = FastAPI(lifespan=lifespan, **kwargs)

    # Add CORS middleware
    if isinstance(settings, AppSettings):
        cors_origins = [origin.strip() for origin in settings.CORS_ORIGINS.split(",")]
        cors_methods = (
            [method.strip() for method in settings.CORS_METHODS.split(",")]
            if settings.CORS_METHODS != "*"
            else ["*"]
        )
        cors_headers = (
            [header.strip() for header in settings.CORS_HEADERS.split(",")]
            if settings.CORS_HEADERS != "*"
            else ["*"]
        )

        app.add_middleware(
            CORSMiddleware,  # type: ignore[arg-type]
            allow_origins=cors_origins,
            allow_credentials=settings.CORS_CREDENTIALS,
            allow_methods=cors_methods,
            allow_headers=cors_headers,
        )

    app.include_router(router)

    add_yaml_middleware_to_router(app, router_path="/api/v1/blueprints")

    return app

1	from contextlib import asynccontextmanager	1✔
2	from typing import Any, AsyncContextManager, AsyncGenerator, Callable	1✔
3
4	import anyio	1✔
5	from arq import create_pool	1✔
6	from arq.connections import RedisSettings	1✔
7	from fastapi import APIRouter, FastAPI	1✔
8	from fastapi.middleware.cors import CORSMiddleware	1✔
9
10	from ..middlewares.yaml_middleware import add_yaml_middleware_to_router	1✔
11	from .config import AppSettings, DatabaseSettings, RedisQueueSettings, settings	1✔
12	from .db.database import Base	1✔
13	from .db.database import async_engine as engine	1✔
14	from .utils import queue	1✔
15
16
17	# Function to create database tables
18	async def create_tables() -> None:	1✔
19	"""Create SQL tables."""
20	async with engine.begin() as conn:	×
21	await conn.run_sync(Base.metadata.create_all, checkfirst=True)	×
22
23
24	async def create_redis_queue_pool() -> None:	1✔
25	"""Create Redis queue pool."""
26	queue.pool = await create_pool(	×
27	RedisSettings(
28	host=settings.REDIS_QUEUE_HOST,
29	port=settings.REDIS_QUEUE_PORT,
30	password=settings.REDIS_QUEUE_PASSWORD,
31	)
32	)
33
34
35	async def close_redis_queue_pool() -> None:	1✔
36	"""Close Redis queue pool."""
37	if queue.pool:	×
38	await queue.pool.aclose()	×
39
40
41	async def set_threadpool_tokens(number_of_tokens: int = 100) -> None:	1✔
42	"""Set thread limit."""
NEW 43	limiter = anyio.to_thread.current_default_thread_limiter() # type: ignore[attr-defined]	×
44	limiter.total_tokens = number_of_tokens	×
45
46
47	# Lifespan factory to manage app lifecycle events
48	def lifespan_factory(	1✔
49	settings: DatabaseSettings \| AppSettings,
50	create_tables_on_start: bool = True,
51	) -> Callable[[FastAPI], AsyncContextManager[Any]]:
52	"""Create a lifespan async context manager for a FastAPI app."""
53
54	@asynccontextmanager	1✔
55	async def lifespan(app: FastAPI) -> AsyncGenerator[Any, None]:	1✔
56	from asyncio import Event # noqa: PLC0415	×
57
58	initialization_complete = Event()	×
59	app.state.initialization_complete = initialization_complete	×
60
61	await set_threadpool_tokens()	×
62
63	try:	×
64	if isinstance(settings, DatabaseSettings) and create_tables_on_start:	×
65	await create_tables()	×
66
67	if isinstance(settings, RedisQueueSettings):	×
68	await create_redis_queue_pool()	×
69
70	initialization_complete.set()	×
71
72	yield	×
73
74	finally:
75	if isinstance(settings, RedisQueueSettings):	×
76	await close_redis_queue_pool()	×
77
78	return lifespan	1✔
79
80
81	# Function to create the FastAPI app
82	def create_application(	1✔
83	router: APIRouter,
84	settings: DatabaseSettings \| AppSettings,
85	create_tables_on_start: bool = True,
86	**kwargs: Any, # noqa: ANN401
87	) -> FastAPI:
88	"""Create and configure a FastAPI application based on the provided settings.
89
90	This function initializes a FastAPI application and configures it with various settings
91	and handlers based on the type of the `settings` object provided.
92
93	Args:
94	----
95	router (APIRouter): The APIRouter object containing the routes to be included in the FastAPI application.
96
97	settings: An instance representing the settings for configuring the FastAPI application.
98	It determines the configuration applied:
99
100	- AppSettings: Configures basic app metadata like name, description, contact, and license info.
101	- DatabaseSettings: Adds event handlers for initializing database tables during startup.
102
103	create_tables_on_start (bool): A flag to indicate whether to create database tables on application startup. Defaults to True.
104
105	**kwargs (Any): Additional keyword arguments passed directly to the FastAPI constructor.
106
107	Returns:
108	-------
109	FastAPI: A fully configured FastAPI application instance.
110
111	The function configures the FastAPI application with different features and behaviors
112	based on the provided settings. It includes setting up database connections, Redis pools
113	for caching, queue, and rate limiting, client-side caching, and customizing the API documentation
114	based on the environment settings.
115
116	"""
117	# --- before creating application ---
118	if isinstance(settings, AppSettings):	1✔
119	to_update = {	1✔
120	"title": settings.APP_NAME,
121	"version": settings.APP_VERSION,
122	"description": settings.APP_DESCRIPTION,
123	"contact": {"name": settings.CONTACT_NAME, "email": settings.CONTACT_EMAIL},
124	"license_info": {
125	"name": settings.LICENSE_NAME,
126	"url": settings.LICENSE_URL,
127	},
128	}
129	kwargs.update(to_update)	1✔
130
131	lifespan = lifespan_factory(settings, create_tables_on_start=create_tables_on_start)	1✔
132
133	app = FastAPI(lifespan=lifespan, **kwargs)	1✔
134
135	# Add CORS middleware
136	if isinstance(settings, AppSettings):	1✔
137	cors_origins = [origin.strip() for origin in settings.CORS_ORIGINS.split(",")]	1✔
138	cors_methods = (	1✔
139	[method.strip() for method in settings.CORS_METHODS.split(",")]
140	if settings.CORS_METHODS != "*"
141	else ["*"]
142	)
143	cors_headers = (	1✔
144	[header.strip() for header in settings.CORS_HEADERS.split(",")]
145	if settings.CORS_HEADERS != "*"
146	else ["*"]
147	)
148
149	app.add_middleware(	1✔
150	CORSMiddleware, # type: ignore[arg-type]
151	allow_origins=cors_origins,
152	allow_credentials=settings.CORS_CREDENTIALS,
153	allow_methods=cors_methods,
154	allow_headers=cors_headers,
155	)
156
157	app.include_router(router)	1✔
158
159	add_yaml_middleware_to_router(app, router_path="/api/v1/blueprints")	1✔
160
161	return app	1✔

OpenLabsHQ / OpenLabs / 20615450373

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous