src/di/factory.py - master - urusai/code_sample

src/di/factory.py
113 строк3 KB
nick
init
30 сен 2024, 16:12
30 сен 2024, 16:12c68451c
import logging
from typing import Literal, Callable

import sentry_sdk
from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import (
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.resources import Resource
from sentry_sdk.integrations.opentelemetry import SentrySpanProcessor

from settings.db.settings import SettingsPostgres
from settings.api.settings import SettingsApplication
from settings.logs.settings import SettingsLogs
from settings.redis.settings import SettingsRedis
from settings.sentry.settings import SentrySettings
from settings.openmetrics.settings import JaegerSettings

logger = logging.getLogger(__name__)


class ConfigFactory:
    def __init__(self):
        self.postgres = SettingsPostgres()
        self.application = SettingsApplication()
        self.redis = SettingsRedis()
        self.logs = SettingsLogs()
        self.sentry = SentrySettings()
        self.jaeger = JaegerSettings()


class EngineFactory:
    ENGINE_TYPES: dict[Literal["sync", "async"], Callable] = {
        "sync": create_engine,
        "async": create_async_engine,
    }

    def __init__(
        self,
        config: SettingsPostgres,
        mode: Literal["sync", "async"],
    ):
        engine_builder = self.ENGINE_TYPES.get(mode)

        if mode == "async":
            self.engine = engine_builder(
                config.dsn(mode=mode),
                future=True,
                echo=False,
                connect_args={"timeout": 0},
                pool_size=0,  # values removed
                max_overflow=0,
                pool_recycle=0,
            )
        else:
            self.engine = engine_builder(
                config.dsn(mode=mode),
            )


class SessionFactory:
    def __init__(self, engine: AsyncEngine):
        self.session_factory = async_sessionmaker(
            engine,
            expire_on_commit=False,
            class_=AsyncSession,
        )


class SentryFactory:
    def __init__(
        self,
        settings: SentrySettings,
        traces_sample_rate: float = 1.0,
    ):
        self.settings = settings
        self.traces_sample_rate = traces_sample_rate

    def init(self):
        sentry_sdk.init(
            dsn=self.settings.DSN,
            traces_sample_rate=self.traces_sample_rate,
            instrumenter="otel",
        )


class TracerFactory:
    def __init__(
        self,
        service_name: str,
        sentry: SentryFactory | None = None,
    ):
        self.service_name = service_name
        self.sentry = sentry

    def __resource(self):
        return Resource(attributes={"service.name": self.service_name})

    def __provider(self) -> TracerProvider:
        return TracerProvider(
            resource=self.__resource(),
        )

    def tracer(self) -> TracerProvider:
        provider = self.__provider()

        if self.sentry:
            provider.add_span_processor(SentrySpanProcessor())
        return provider