sh-edraft.de/cpl
5
0
Fork 0
Code Issues 5 Pull Requests 2 Actions Packages Projects 1 Releases 12 Wiki Activity

Restructuring
All checks were successful
Build on push / prepare (push) Successful in 8s
Details
Build on push / query (push) Successful in 17s
Details
Build on push / core (push) Successful in 26s
Details
Build on push / translation (push) Successful in 15s
Details
Build on push / mail (push) Successful in 17s
Details

Browse Source
This commit is contained in:
Sven Heidemann
2025-09-16 08:48:07 +02:00
parent 5f25400bcd
commit b97bc0a3ed
72 changed files with 261 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/cpl-core/cpl/core/application/__init__.py
View File

@@ -1,6 +0,0 @@
from .application_abc import ApplicationABC
from .application_builder import ApplicationBuilder
from .application_builder_abc import ApplicationBuilderABC
from .application_extension_abc import ApplicationExtensionABC
from .startup_abc import StartupABC
from .startup_extension_abc import StartupExtensionABC

58
src/cpl-core/cpl/core/application/application_abc.py
View File

@@ -1,58 +0,0 @@
from abc import ABC, abstractmethod
from typing import Optional
from cpl.core.configuration import Configuration
from cpl.core.console.console import Console
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.core.environment import Environment
class ApplicationABC(ABC):
r"""ABC for the Application class
Parameters:
config: :class:`cpl.core.configuration.configuration_abc.ConfigurationABC`
Contains object loaded from appsettings
services: :class:`cpl.core.dependency_injection.service_provider_abc.ServiceProviderABC`
Contains instances of prepared objects
"""
@abstractmethod
def __init__(self, services: ServiceProviderABC):
self._services: Optional[ServiceProviderABC] = services
def run(self):
r"""Entry point
Called by custom Application.main
"""
try:
self.configure()
self.main()
except KeyboardInterrupt:
Console.close()
async def run_async(self):
r"""Entry point
Called by custom Application.main
"""
try:
await self.configure()
await self.main()
except KeyboardInterrupt:
Console.close()
@abstractmethod
def configure(self):
r"""Configure the application
Called by :class:`cpl.core.application.application_abc.ApplicationABC.run`
"""
@abstractmethod
def main(self):
r"""Custom entry point
Called by :class:`cpl.core.application.application_abc.ApplicationABC.run`
"""

97
src/cpl-core/cpl/core/application/application_builder.py
View File

@@ -1,97 +0,0 @@
from typing import Type, Optional, Callable, Union
from cpl.core.application.application_abc import ApplicationABC
from cpl.core.application.application_builder_abc import ApplicationBuilderABC
from cpl.core.application.application_extension_abc import ApplicationExtensionABC
from cpl.core.application.async_application_extension_abc import AsyncApplicationExtensionABC
from cpl.core.application.async_startup_abc import AsyncStartupABC
from cpl.core.application.async_startup_extension_abc import AsyncStartupExtensionABC
from cpl.core.application.startup_abc import StartupABC
from cpl.core.application.startup_extension_abc import StartupExtensionABC
from cpl.core.configuration.configuration import Configuration
from cpl.core.dependency_injection.service_collection import ServiceCollection
from cpl.core.environment import Environment
class ApplicationBuilder(ApplicationBuilderABC):
r"""This is class is used to build an object of :class:`cpl.core.application.application_abc.ApplicationABC`
Parameter:
app: Type[:class:`cpl.core.application.application_abc.ApplicationABC`]
Application to build
"""
def __init__(self, app: Type[ApplicationABC]):
ApplicationBuilderABC.__init__(self)
self._app = app
self._startup: Optional[StartupABC | AsyncStartupABC] = None
self._services = ServiceCollection()
self._app_extensions: list[Type[ApplicationExtensionABC | AsyncApplicationExtensionABC]] = []
self._startup_extensions: list[Type[StartupExtensionABC | AsyncStartupABC]] = []
def use_startup(self, startup: Type[StartupABC | AsyncStartupABC]) -> "ApplicationBuilder":
self._startup = startup()
return self
def use_extension(
self,
extension: Type[
ApplicationExtensionABC | AsyncApplicationExtensionABC | StartupExtensionABC | AsyncStartupExtensionABC
],
) -> "ApplicationBuilder":
if (
issubclass(extension, ApplicationExtensionABC) or issubclass(extension, AsyncApplicationExtensionABC)
) and extension not in self._app_extensions:
self._app_extensions.append(extension)
elif (
issubclass(extension, StartupExtensionABC) or issubclass(extension, AsyncStartupExtensionABC)
) and extension not in self._startup_extensions:
self._startup_extensions.append(extension)
return self
def _build_startup(self):
for ex in self._startup_extensions:
extension = ex()
extension.configure_configuration(Configuration, Environment)
extension.configure_services(self._services, Environment)
if self._startup is not None:
self._startup.configure_configuration(Configuration, Environment)
self._startup.configure_services(self._services, Environment)
async def _build_async_startup(self):
for ex in self._startup_extensions:
extension = ex()
await extension.configure_configuration(Configuration, Environment)
await extension.configure_services(self._services, Environment)
if self._startup is not None:
await self._startup.configure_configuration(Configuration, Environment)
await self._startup.configure_services(self._services, Environment)
def build(self) -> ApplicationABC:
self._build_startup()
config = Configuration
services = self._services.build_service_provider()
for ex in self._app_extensions:
extension = ex()
extension.run(config, services)
return self._app(services)
async def build_async(self) -> ApplicationABC:
await self._build_async_startup()
config = Configuration
services = self._services.build_service_provider()
for ex in self._app_extensions:
extension = ex()
await extension.run(config, services)
return self._app(services)

47
src/cpl-core/cpl/core/application/application_builder_abc.py
View File

@@ -1,47 +0,0 @@
from abc import ABC, abstractmethod
from typing import Type
from cpl.core.application.application_abc import ApplicationABC
from cpl.core.application.startup_abc import StartupABC
class ApplicationBuilderABC(ABC):
r"""ABC for the :class:`cpl.core.application.application_builder.ApplicationBuilder`"""
@abstractmethod
def __init__(self, *args):
pass
@abstractmethod
def use_startup(self, startup: Type[StartupABC]):
r"""Sets the custom startup class to use
Parameter:
startup: Type[:class:`cpl.core.application.startup_abc.StartupABC`]
Startup class to use
"""
@abstractmethod
async def use_startup(self, startup: Type[StartupABC]):
r"""Sets the custom startup class to use async
Parameter:
startup: Type[:class:`cpl.core.application.startup_abc.StartupABC`]
Startup class to use
"""
@abstractmethod
def build(self) -> ApplicationABC:
r"""Creates custom application object
Returns:
Object of :class:`cpl.core.application.application_abc.ApplicationABC`
"""
@abstractmethod
async def build_async(self) -> ApplicationABC:
r"""Creates custom application object async
Returns:
Object of :class:`cpl.core.application.application_abc.ApplicationABC`
"""

14
src/cpl-core/cpl/core/application/application_extension_abc.py
View File

@@ -1,14 +0,0 @@
from abc import ABC, abstractmethod
from cpl.core.configuration.configuration import Configuration
from cpl.core.dependency_injection import ServiceProviderABC
class ApplicationExtensionABC(ABC):
@abstractmethod
def __init__(self):
pass
@abstractmethod
def run(self, config: Configuration, services: ServiceProviderABC):
pass

14
src/cpl-core/cpl/core/application/async_application_extension_abc.py
View File

@@ -1,14 +0,0 @@
from abc import ABC, abstractmethod
from cpl.core.configuration.configuration import Configuration
from cpl.core.dependency_injection import ServiceProviderABC
class AsyncApplicationExtensionABC(ABC):
@abstractmethod
def __init__(self):
pass
@abstractmethod
async def run(self, config: Configuration, services: ServiceProviderABC):
pass

23
src/cpl-core/cpl/core/application/async_startup_abc.py
View File

@@ -1,23 +0,0 @@
from abc import ABC, abstractmethod
from cpl.core.dependency_injection.service_collection_abc import ServiceCollectionABC
class AsyncStartupABC(ABC):
r"""ABC for the startup class"""
@abstractmethod
def __init__(self):
pass
@abstractmethod
async def configure_configuration(self):
r"""Creates configuration of application"""
@abstractmethod
async def configure_services(self, service: ServiceCollectionABC):
r"""Creates service provider
Parameter:
services: :class:`cpl.core.dependency_injection.service_collection_abc`
"""

31
src/cpl-core/cpl/core/application/async_startup_extension_abc.py
View File

@@ -1,31 +0,0 @@
from abc import ABC, abstractmethod
from cpl.core.configuration.configuration import Configuration
from cpl.core.dependency_injection.service_collection_abc import ServiceCollectionABC
from cpl.core.environment.environment import Environment
class AsyncStartupExtensionABC(ABC):
r"""ABC for startup extension classes"""
@abstractmethod
def __init__(self):
pass
@abstractmethod
async def configure_configuration(self, config: Configuration, env: Environment):
r"""Creates configuration of application
Parameter:
config: :class:`cpl.core.configuration.configuration_abc.Configuration`
env: :class:`cpl.core.environment.application_environment_abc`
"""
@abstractmethod
async def configure_services(self, service: ServiceCollectionABC, env: Environment):
r"""Creates service provider
Parameter:
services: :class:`cpl.core.dependency_injection.service_collection_abc`
env: :class:`cpl.core.environment.application_environment_abc`
"""

31
src/cpl-core/cpl/core/application/startup_abc.py
View File

@@ -1,31 +0,0 @@
from abc import ABC, abstractmethod
from cpl.core.configuration import Configuration
from cpl.core.dependency_injection.service_collection_abc import ServiceCollectionABC
from cpl.core.environment import Environment
class StartupABC(ABC):
r"""ABC for the startup class"""
@abstractmethod
def __init__(self):
pass
@abstractmethod
def configure_configuration(self, config: Configuration, env: Environment):
r"""Creates configuration of application
Parameter:
config: :class:`cpl.core.configuration.configuration_abc.ConfigurationABC`
env: :class:`cpl.core.environment.application_environment_abc`
"""
@abstractmethod
def configure_services(self, service: ServiceCollectionABC, env: Environment):
r"""Creates service provider
Parameter:
services: :class:`cpl.core.dependency_injection.service_collection_abc`
env: :class:`cpl.core.environment.application_environment_abc`
"""

33
src/cpl-core/cpl/core/application/startup_extension_abc.py
View File

@@ -1,33 +0,0 @@
from abc import ABC, abstractmethod
from cpl.core.configuration import Configuration
from cpl.core.dependency_injection.service_collection_abc import ServiceCollectionABC
from cpl.core.environment.environment import Environment
class StartupExtensionABC(ABC):
r"""ABC for startup extension classes"""
@abstractmethod
def __init__(self):
pass
@abstractmethod
def configure_configuration(self, config: Configuration, env: Environment):
r"""Creates configuration of application
Parameter:
config: :class:`cpl.core.configuration.configuration_abc.ConfigurationABC`
env: :class:`cpl.core.environment.application_environment_abc`
"""
@abstractmethod
def configure_services(self, service: ServiceCollectionABC, env: Environment):
r"""Creates service provider
Parameter:
services: :class:`cpl.core.dependency_injection.service_collection_abc`
env: :class:`cpl.core.environment.application_environment_abc`
"""

3
src/cpl-core/cpl/core/database/__init__.py
View File

@@ -1,3 +0,0 @@
from .database_settings_name_enum import DatabaseSettingsNameEnum
from .database_settings import DatabaseSettings
from .table_abc import TableABC

2
src/cpl-core/cpl/core/database/connection/__init__.py
View File

@@ -1,2 +0,0 @@
from .database_connection import DatabaseConnection
from .database_connection_abc import DatabaseConnectionABC

54
src/cpl-core/cpl/core/database/connection/database_connection.py
View File

@@ -1,54 +0,0 @@
from typing import Optional
import mysql.connector as sql
from mysql.connector.abstracts import MySQLConnectionAbstract
from mysql.connector.cursor import MySQLCursorBuffered
from cpl.core.database.connection.database_connection_abc import DatabaseConnectionABC
from cpl.core.database.database_settings import DatabaseSettings
from cpl.core.utils.credential_manager import CredentialManager
class DatabaseConnection(DatabaseConnectionABC):
r"""Representation of the database connection"""
def __init__(self):
DatabaseConnectionABC.__init__(self)
self._database: Optional[MySQLConnectionAbstract] = None
self._cursor: Optional[MySQLCursorBuffered] = None
@property
def server(self) -> MySQLConnectionAbstract:
return self._database
@property
def cursor(self) -> MySQLCursorBuffered:
return self._cursor
def connect(self, settings: DatabaseSettings):
connection = sql.connect(
host=settings.host,
port=settings.port,
user=settings.user,
passwd=CredentialManager.decrypt(settings.password),
charset=settings.charset,
use_unicode=settings.use_unicode,
buffered=settings.buffered,
auth_plugin=settings.auth_plugin,
ssl_disabled=settings.ssl_disabled,
)
connection.cursor().execute(f"CREATE DATABASE IF NOT EXISTS `{settings.database}`;")
self._database = sql.connect(
host=settings.host,
port=settings.port,
user=settings.user,
passwd=CredentialManager.decrypt(settings.password),
db=settings.database,
charset=settings.charset,
use_unicode=settings.use_unicode,
buffered=settings.buffered,
auth_plugin=settings.auth_plugin,
ssl_disabled=settings.ssl_disabled,
)
self._cursor = self._database.cursor()

32
src/cpl-core/cpl/core/database/connection/database_connection_abc.py
View File

@@ -1,32 +0,0 @@
from abc import ABC, abstractmethod
from cpl.core.database.database_settings import DatabaseSettings
from mysql.connector.abstracts import MySQLConnectionAbstract
from mysql.connector.cursor import MySQLCursorBuffered
class DatabaseConnectionABC(ABC):
r"""ABC for the :class:`cpl.core.database.connection.database_connection.DatabaseConnection`"""
@abstractmethod
def __init__(self):
pass
@property
@abstractmethod
def server(self) -> MySQLConnectionAbstract:
pass
@property
@abstractmethod
def cursor(self) -> MySQLCursorBuffered:
pass
@abstractmethod
def connect(self, database_settings: DatabaseSettings):
r"""Connects to a database by connection string
Parameter:
connection_string: :class:`str`
Database connection string, see: https://docs.sqlalchemy.org/en/14/core/engines.html
"""

2
src/cpl-core/cpl/core/database/context/__init__.py
View File

@@ -1,2 +0,0 @@
from .database_context import DatabaseContext
from .database_context_abc import DatabaseContextABC

52
src/cpl-core/cpl/core/database/context/database_context.py
View File

@@ -1,52 +0,0 @@
from typing import Optional
from cpl.core.database.connection.database_connection import DatabaseConnection
from cpl.core.database.connection.database_connection_abc import DatabaseConnectionABC
from cpl.core.database.context.database_context_abc import DatabaseContextABC
from cpl.core.database.database_settings import DatabaseSettings
from mysql.connector.cursor import MySQLCursorBuffered
class DatabaseContext(DatabaseContextABC):
r"""Representation of the database context
Parameter:
database_settings: :class:`cpl.core.database.database_settings.DatabaseSettings`
"""
def __init__(self):
DatabaseContextABC.__init__(self)
self._db: DatabaseConnectionABC = DatabaseConnection()
self._settings: Optional[DatabaseSettings] = None
@property
def cursor(self) -> MySQLCursorBuffered:
self._ping_and_reconnect()
return self._db.cursor
def _ping_and_reconnect(self):
try:
self._db.server.ping(reconnect=True, attempts=3, delay=5)
except Exception:
# reconnect your cursor as you did in __init__ or wherever
if self._settings is None:
raise Exception("Call DatabaseContext.connect first")
self.connect(self._settings)
def connect(self, database_settings: DatabaseSettings):
if self._settings is None:
self._settings = database_settings
self._db.connect(database_settings)
self.save_changes()
def save_changes(self):
self._ping_and_reconnect()
self._db.server.commit()
def select(self, statement: str) -> list[tuple]:
self._ping_and_reconnect()
self._db.cursor.execute(statement)
return self._db.cursor.fetchall()

40
src/cpl-core/cpl/core/database/context/database_context_abc.py
View File

@@ -1,40 +0,0 @@
from abc import ABC, abstractmethod
from cpl.core.database.database_settings import DatabaseSettings
from mysql.connector.cursor import MySQLCursorBuffered
class DatabaseContextABC(ABC):
r"""ABC for the :class:`cpl.core.database.context.database_context.DatabaseContext`"""
@abstractmethod
def __init__(self, *args):
pass
@property
@abstractmethod
def cursor(self) -> MySQLCursorBuffered:
pass
@abstractmethod
def connect(self, database_settings: DatabaseSettings):
r"""Connects to a database by connection settings
Parameter:
database_settings :class:`cpl.core.database.database_settings.DatabaseSettings`
"""
@abstractmethod
def save_changes(self):
r"""Saves changes of the database"""
@abstractmethod
def select(self, statement: str) -> list[tuple]:
r"""Runs SQL Statements
Parameter:
statement: :class:`str`
Returns:
list: Fetched list of selected elements
"""

73
src/cpl-core/cpl/core/database/database_settings.py
View File

@@ -1,73 +0,0 @@
from typing import Optional
from cpl.core.configuration.configuration_model_abc import ConfigurationModelABC
class DatabaseSettings(ConfigurationModelABC):
r"""Represents settings for the database connection"""
def __init__(
self,
host: str = None,
port: int = 3306,
user: str = None,
password: str = None,
database: str = None,
charset: str = "utf8mb4",
use_unicode: bool = False,
buffered: bool = False,
auth_plugin: str = "caching_sha2_password",
ssl_disabled: bool = False,
):
ConfigurationModelABC.__init__(self)
self._host: Optional[str] = host
self._port: Optional[int] = port
self._user: Optional[str] = user
self._password: Optional[str] = password
self._database: Optional[str] = database
self._charset: Optional[str] = charset
self._use_unicode: Optional[bool] = use_unicode
self._buffered: Optional[bool] = buffered
self._auth_plugin: Optional[str] = auth_plugin
self._ssl_disabled: Optional[bool] = ssl_disabled
@property
def host(self) -> Optional[str]:
return self._host
@property
def port(self) -> Optional[int]:
return self._port
@property
def user(self) -> Optional[str]:
return self._user
@property
def password(self) -> Optional[str]:
return self._password
@property
def database(self) -> Optional[str]:
return self._database
@property
def charset(self) -> Optional[str]:
return self._charset
@property
def use_unicode(self) -> Optional[bool]:
return self._use_unicode
@property
def buffered(self) -> Optional[bool]:
return self._buffered
@property
def auth_plugin(self) -> Optional[str]:
return self._auth_plugin
@property
def ssl_disabled(self) -> Optional[bool]:
return self._ssl_disabled

13
src/cpl-core/cpl/core/database/database_settings_name_enum.py
View File

@@ -1,13 +0,0 @@
from enum import Enum
class DatabaseSettingsNameEnum(Enum):
host = "Host"
port = "Port"
user = "User"
password = "Password"
database = "Database"
charset = "Charset"
use_unicode = "UseUnicode"
buffered = "Buffered"
auth_plugin = "AuthPlugin"

8
src/cpl-core/cpl/core/database/db_logger.py
View File

@@ -1,8 +0,0 @@
from cpl.core.log import Logger
from cpl.core.typing import Source
class DBLogger(Logger):
def __init__(self, source: Source):
Logger.__init__(self, source, "db")

37
src/cpl-core/cpl/core/database/table_abc.py
View File

@@ -1,37 +0,0 @@
from abc import ABC, abstractmethod
from datetime import datetime
from typing import Optional
class TableABC(ABC):
@abstractmethod
def __init__(self):
self._created_at: Optional[datetime] = datetime.now().isoformat()
self._modified_at: Optional[datetime] = datetime.now().isoformat()
@property
def created_at(self) -> datetime:
return self._created_at
@property
def modified_at(self) -> datetime:
return self._modified_at
@modified_at.setter
def modified_at(self, value: datetime):
self._modified_at = value
@property
@abstractmethod
def insert_string(self) -> str:
pass
@property
@abstractmethod
def udpate_string(self) -> str:
pass
@property
@abstractmethod
def delete_string(self) -> str:
pass

8
src/cpl-core/cpl/core/dependency_injection/__init__.py
View File

@@ -1,8 +0,0 @@
from .scope import Scope
from .scope_abc import ScopeABC
from .service_collection import ServiceCollection
from .service_collection_abc import ServiceCollectionABC
from .service_descriptor import ServiceDescriptor
from .service_lifetime_enum import ServiceLifetimeEnum
from .service_provider import ServiceProvider
from .service_provider_abc import ServiceProviderABC

22
src/cpl-core/cpl/core/dependency_injection/scope.py
View File

@@ -1,22 +0,0 @@
from cpl.core.dependency_injection.scope_abc import ScopeABC
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
class Scope(ScopeABC):
def __init__(self, service_provider: ServiceProviderABC):
self._service_provider = service_provider
self._service_provider.set_scope(self)
ScopeABC.__init__(self)
def __enter__(self):
return self
def __exit__(self, *args):
self.dispose()
@property
def service_provider(self) -> ServiceProviderABC:
return self._service_provider
def dispose(self):
self._service_provider = None

21
src/cpl-core/cpl/core/dependency_injection/scope_abc.py
View File

@@ -1,21 +0,0 @@
from abc import ABC, abstractmethod
class ScopeABC(ABC):
r"""ABC for the class :class:`cpl.core.dependency_injection.scope.Scope`"""
def __init__(self):
pass
@property
@abstractmethod
def service_provider(self):
r"""Returns to service provider of scope
Returns:
Object of type :class:`cpl.core.dependency_injection.service_provider_abc.ServiceProviderABC`
"""
@abstractmethod
def dispose(self):
r"""Sets service_provider to None"""

18
src/cpl-core/cpl/core/dependency_injection/scope_builder.py
View File

@@ -1,18 +0,0 @@
from cpl.core.dependency_injection.scope import Scope
from cpl.core.dependency_injection.scope_abc import ScopeABC
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
class ScopeBuilder:
r"""Class to build :class:`cpl.core.dependency_injection.scope.Scope`"""
def __init__(self, service_provider: ServiceProviderABC) -> None:
self._service_provider = service_provider
def build(self) -> ScopeABC:
r"""Returns scope
Returns:
Object of type :class:`cpl.core.dependency_injection.scope.Scope`
"""
return Scope(self._service_provider)

76
src/cpl-core/cpl/core/dependency_injection/service_collection.py
View File

@@ -1,76 +0,0 @@
from typing import Union, Type, Callable, Optional
from cpl.core.database.context.database_context_abc import DatabaseContextABC
from cpl.core.database.database_settings import DatabaseSettings
from cpl.core.dependency_injection.service_collection_abc import ServiceCollectionABC
from cpl.core.dependency_injection.service_descriptor import ServiceDescriptor
from cpl.core.dependency_injection.service_lifetime_enum import ServiceLifetimeEnum
from cpl.core.dependency_injection.service_provider import ServiceProvider
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.core.log.logger import Logger
from cpl.core.log.logger_abc import LoggerABC
from cpl.core.pipes.pipe_abc import PipeABC
from cpl.core.typing import T, Service
class ServiceCollection(ServiceCollectionABC):
r"""Representation of the collection of services"""
def __init__(self):
ServiceCollectionABC.__init__(self)
self._database_context: Optional[DatabaseContextABC] = None
self._service_descriptors: list[ServiceDescriptor] = []
def _add_descriptor(self, service: Union[type, object], lifetime: ServiceLifetimeEnum, base_type: Callable = None):
found = False
for descriptor in self._service_descriptors:
if isinstance(service, descriptor.service_type):
found = True
if found:
service_type = service
if not isinstance(service, type):
service_type = type(service)
raise Exception(f"Service of type {service_type} already exists")
self._service_descriptors.append(ServiceDescriptor(service, lifetime, base_type))
def _add_descriptor_by_lifetime(self, service_type: Type, lifetime: ServiceLifetimeEnum, service: Callable = None):
if service is not None:
self._add_descriptor(service, lifetime, service_type)
else:
self._add_descriptor(service_type, lifetime)
return self
def add_db_context(self, db_context_type: Type[DatabaseContextABC], db_settings: DatabaseSettings):
self.add_singleton(DatabaseContextABC, db_context_type)
self._database_context = self.build_service_provider().get_service(DatabaseContextABC)
self._database_context.connect(db_settings)
def add_logging(self):
self.add_transient(LoggerABC, Logger)
return self
def add_pipes(self):
for pipe in PipeABC.__subclasses__():
self.add_transient(PipeABC, pipe)
return self
def add_singleton(self, service_type: T, service: Service = None):
self._add_descriptor_by_lifetime(service_type, ServiceLifetimeEnum.singleton, service)
return self
def add_scoped(self, service_type: T, service: Service = None):
self._add_descriptor_by_lifetime(service_type, ServiceLifetimeEnum.scoped, service)
return self
def add_transient(self, service_type: T, service: Service = None):
self._add_descriptor_by_lifetime(service_type, ServiceLifetimeEnum.transient, service)
return self
def build_service_provider(self) -> ServiceProviderABC:
sp = ServiceProvider(self._service_descriptors, self._database_context)
ServiceProviderABC.set_global_provider(sp)
return sp

90
src/cpl-core/cpl/core/dependency_injection/service_collection_abc.py
View File

@@ -1,90 +0,0 @@
from abc import abstractmethod, ABC
from typing import Type
from cpl.core.database.context.database_context_abc import DatabaseContextABC
from cpl.core.database.database_settings import DatabaseSettings
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.core.typing import T, Source
class ServiceCollectionABC(ABC):
r"""ABC for the class :class:`cpl.core.dependency_injection.service_collection.ServiceCollection`"""
@abstractmethod
def __init__(self):
pass
@abstractmethod
def add_db_context(self, db_context_type: Type[DatabaseContextABC], db_settings: DatabaseSettings):
r"""Adds database context
Parameter:
db_context: Type[:class:`cpl.core.database.context.database_context_abc.DatabaseContextABC`]
Database context
"""
@abstractmethod
def add_logging(self):
r"""Adds the CPL internal logger"""
@abstractmethod
def add_pipes(self):
r"""Adds the CPL internal pipes as transient"""
def add_translation(self):
r"""Adds the CPL translation"""
raise NotImplementedError("You should install and use the cpl-translation package")
def add_mail(self):
r"""Adds the CPL mail"""
raise NotImplementedError("You should install and use the cpl-mail package")
@abstractmethod
def add_transient(self, service_type: T, service: T = None) -> "ServiceCollectionABC":
r"""Adds a service with transient lifetime
Parameter:
service_type: :class:`Type`
Type of the service
service: :class:`Callable`
Object of the service
Returns:
self: :class:`cpl.core.dependency_injection.service_collection_abc.ServiceCollectionABC`
"""
@abstractmethod
def add_scoped(self, service_type: T, service: T = None) -> "ServiceCollectionABC":
r"""Adds a service with scoped lifetime
Parameter:
service_type: :class:`Type`
Type of the service
service: :class:`Callable`
Object of the service
Returns:
self: :class:`cpl.core.dependency_injection.service_collection_abc.ServiceCollectionABC`
"""
@abstractmethod
def add_singleton(self, service_type: T, service: T = None) -> "ServiceCollectionABC":
r"""Adds a service with singleton lifetime
Parameter:
service_type: :class:`Type`
Type of the service
service: :class:`Callable`
Object of the service
Returns:
self: :class:`cpl.core.dependency_injection.service_collection_abc.ServiceCollectionABC`
"""
@abstractmethod
def build_service_provider(self) -> ServiceProviderABC:
r"""Creates instance of the service provider
Returns:
Object of type :class:`cpl.core.dependency_injection.service_provider_abc.ServiceProviderABC`
"""

46
src/cpl-core/cpl/core/dependency_injection/service_descriptor.py
View File

@@ -1,46 +0,0 @@
from typing import Union, Optional
from cpl.core.dependency_injection.service_lifetime_enum import ServiceLifetimeEnum
class ServiceDescriptor:
r"""Descriptor of a service
Parameter:
implementation: Union[:class:`type`, Optional[:class:`object`]]
Object or type of service
lifetime: :class:`cpl.core.dependency_injection.service_lifetime_enum.ServiceLifetimeEnum`
Lifetime of the service
"""
def __init__(self, implementation: Union[type, Optional[object]], lifetime: ServiceLifetimeEnum, base_type=None):
self._service_type = implementation
self._implementation = implementation
self._lifetime = lifetime
if not isinstance(implementation, type):
self._service_type = type(implementation)
else:
self._implementation = None
self._base_type = base_type if base_type is not None else self._service_type
@property
def service_type(self) -> type:
return self._service_type
@property
def base_type(self) -> type:
return self._base_type
@property
def implementation(self) -> Union[type, Optional[object]]:
return self._implementation
@implementation.setter
def implementation(self, implementation: Union[type, Optional[object]]):
self._implementation = implementation
@property
def lifetime(self) -> ServiceLifetimeEnum:
return self._lifetime

7
src/cpl-core/cpl/core/dependency_injection/service_lifetime_enum.py
View File

@@ -1,7 +0,0 @@
from enum import Enum
class ServiceLifetimeEnum(Enum):
singleton = 0
scoped = 1
transient = 2

172
src/cpl-core/cpl/core/dependency_injection/service_provider.py
View File

@@ -1,172 +0,0 @@
import copy
import typing
from inspect import signature, Parameter, Signature
from typing import Optional
from cpl.core.configuration import Configuration
from cpl.core.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.core.database.context.database_context_abc import DatabaseContextABC
from cpl.core.dependency_injection.scope_abc import ScopeABC
from cpl.core.dependency_injection.scope_builder import ScopeBuilder
from cpl.core.dependency_injection.service_descriptor import ServiceDescriptor
from cpl.core.dependency_injection.service_lifetime_enum import ServiceLifetimeEnum
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.core.environment import Environment
from cpl.core.typing import T, R, Source
class ServiceProvider(ServiceProviderABC):
r"""Provider for the services
Parameter
---------
service_descriptors: list[:class:`cpl.core.dependency_injection.service_descriptor.ServiceDescriptor`]
Descriptor of the service
config: :class:`cpl.core.configuration.configuration_abc.ConfigurationABC`
CPL Configuration
db_context: Optional[:class:`cpl.core.database.context.database_context_abc.DatabaseContextABC`]
Database representation
"""
def __init__(
self,
service_descriptors: list[ServiceDescriptor],
db_context: Optional[DatabaseContextABC],
):
ServiceProviderABC.__init__(self)
self._service_descriptors: list[ServiceDescriptor] = service_descriptors
self._database_context = db_context
self._scope: Optional[ScopeABC] = None
def _find_service(self, service_type: type) -> Optional[ServiceDescriptor]:
for descriptor in self._service_descriptors:
if descriptor.service_type == service_type or issubclass(descriptor.base_type, service_type):
return descriptor
return None
def _get_service(self, parameter: Parameter, origin_service_type: type = None) -> Optional[object]:
for descriptor in self._service_descriptors:
if descriptor.service_type == parameter.annotation or issubclass(
descriptor.service_type, parameter.annotation
):
if descriptor.implementation is not None:
return descriptor.implementation
implementation = self._build_service(descriptor.service_type, origin_service_type=origin_service_type)
if descriptor.lifetime == ServiceLifetimeEnum.singleton:
descriptor.implementation = implementation
return implementation
# raise Exception(f'Service {parameter.annotation} not found')
def _get_services(self, t: type, *args, service_type: type = None, **kwargs) -> list[Optional[object]]:
implementations = []
for descriptor in self._service_descriptors:
if descriptor.service_type == t or issubclass(descriptor.service_type, t):
if descriptor.implementation is not None:
implementations.append(descriptor.implementation)
continue
implementation = self._build_service(descriptor.service_type, *args, service_type, **kwargs)
if descriptor.lifetime == ServiceLifetimeEnum.singleton:
descriptor.implementation = implementation
implementations.append(implementation)
return implementations
def _build_by_signature(self, sig: Signature, origin_service_type: type) -> list[R]:
params = []
for param in sig.parameters.items():
parameter = param[1]
if parameter.name != "self" and parameter.annotation != Parameter.empty:
if typing.get_origin(parameter.annotation) == list:
params.append(self._get_services(typing.get_args(parameter.annotation)[0], origin_service_type))
elif parameter.annotation == Source:
params.append(origin_service_type.__name__)
elif issubclass(parameter.annotation, ServiceProviderABC):
params.append(self)
elif issubclass(parameter.annotation, Environment):
params.append(Environment)
elif issubclass(parameter.annotation, DatabaseContextABC):
params.append(self._database_context)
elif issubclass(parameter.annotation, ConfigurationModelABC):
params.append(Configuration.get(parameter.annotation))
elif issubclass(parameter.annotation, Configuration):
params.append(Configuration)
else:
params.append(self._get_service(parameter, origin_service_type))
return params
def _build_service(self, service_type: type, *args, origin_service_type: type = None, **kwargs) -> object:
if origin_service_type is None:
origin_service_type = service_type
for descriptor in self._service_descriptors:
if descriptor.service_type == service_type or issubclass(descriptor.service_type, service_type):
if descriptor.implementation is not None:
service_type = type(descriptor.implementation)
else:
service_type = descriptor.service_type
break
sig = signature(service_type.__init__)
params = self._build_by_signature(sig, origin_service_type)
return service_type(*params, *args, **kwargs)
def set_scope(self, scope: ScopeABC):
self._scope = scope
def create_scope(self) -> ScopeABC:
descriptors = []
for descriptor in self._service_descriptors:
if descriptor.lifetime == ServiceLifetimeEnum.singleton:
descriptors.append(descriptor)
else:
descriptors.append(copy.deepcopy(descriptor))
sb = ScopeBuilder(ServiceProvider(descriptors, self._database_context))
return sb.build()
def get_service(self, service_type: T, *args, **kwargs) -> Optional[R]:
result = self._find_service(service_type)
if result is None:
return None
if result.implementation is not None:
return result.implementation
implementation = self._build_service(service_type, *args, **kwargs)
if (
result.lifetime == ServiceLifetimeEnum.singleton
or result.lifetime == ServiceLifetimeEnum.scoped
and self._scope is not None
):
result.implementation = implementation
return implementation
def get_services(self, service_type: T, *args, **kwargs) -> list[Optional[R]]:
implementations = []
if typing.get_origin(service_type) == list:
raise Exception(f"Invalid type {service_type}! Expected single type not list of type")
implementations.extend(self._get_services(service_type))
return implementations

111
src/cpl-core/cpl/core/dependency_injection/service_provider_abc.py
View File

@@ -1,111 +0,0 @@
import functools
from abc import abstractmethod, ABC
from inspect import Signature, signature
from typing import Optional
from cpl.core.dependency_injection.scope_abc import ScopeABC
from cpl.core.typing import T, R
class ServiceProviderABC(ABC):
r"""ABC for the class :class:`cpl.core.dependency_injection.service_provider.ServiceProvider`"""
_provider: Optional["ServiceProviderABC"] = None
@abstractmethod
def __init__(self):
pass
@classmethod
def set_global_provider(cls, provider: "ServiceProviderABC"):
cls._provider = provider
@abstractmethod
def _build_by_signature(self, sig: Signature, origin_service_type: type) -> list[R]:
pass
@abstractmethod
def _build_service(self, service_type: type, *args, **kwargs) -> object:
r"""Creates instance of given type
Parameter
---------
instance_type: :class:`type`
The type of the searched instance
Returns
-------
Object of the given type
"""
@abstractmethod
def set_scope(self, scope: ScopeABC):
r"""Sets the scope of service provider
Parameter
---------
Object of type :class:`cpl.core.dependency_injection.scope_abc.ScopeABC`
Service scope
"""
@abstractmethod
def create_scope(self) -> ScopeABC:
r"""Creates a service scope
Returns
-------
Object of type :class:`cpl.core.dependency_injection.scope_abc.ScopeABC`
"""
@abstractmethod
def get_service(self, instance_type: T, *args, **kwargs) -> Optional[R]:
r"""Returns instance of given type
Parameter
---------
instance_type: :class:`cpl.core.type.T`
The type of the searched instance
Returns
-------
Object of type Optional[:class:`cpl.core.type.T`]
"""
@abstractmethod
def get_services(self, service_type: T, *args, **kwargs) -> list[Optional[R]]:
r"""Returns instance of given type
Parameter
---------
service_type: :class:`cpl.core.type.T`
The type of the searched instance
Returns
-------
Object of type list[Optional[:class:`cpl.core.type.T`]
"""
@classmethod
def inject(cls, f=None):
r"""Decorator to allow injection into static and class methods
Parameter
---------
f: Callable
Returns
-------
function
"""
if f is None:
return functools.partial(cls.inject)
@functools.wraps(f)
def inner(*args, **kwargs):
if cls._provider is None:
raise Exception(f"{cls.__name__} not build!")
injection = [x for x in cls._provider._build_by_signature(signature(f)) if x is not None]
return f(*args, *injection, **kwargs)
return inner