Repository URL to install this package:
|
Version:
3.0.3 ▾
|
sincpro-framework
/
use_bus.pyi
|
|---|
from typing import Any, Callable, Dict, Type, TypeVar, overload
from _typeshed import Incomplete
from sincpro_log.logger import LoggerProxy
from typing_extensions import Self
from . import ioc as ioc
from .bus import FrameworkBus as FrameworkBus
from .context.framework_context import FrameworkContext
from .context.mixin import ContextMixin
from .exceptions import DependencyAlreadyRegistered as DependencyAlreadyRegistered
from .exceptions import SincproFrameworkNotBuilt as SincproFrameworkNotBuilt
from .middleware import Middleware, MiddlewarePipeline
from .sincpro_abstractions import ApplicationService, DataTransferObject, Feature
from .sincpro_abstractions import TypeDTO as TypeDTO
from .sincpro_abstractions import TypeDTOResponse as TypeDTOResponse
from .sincpro_logger import create_logger as create_logger
# Type alias for decorator functions
DecoratorFunction = Callable[[Type], Type]
class UseFramework(ContextMixin):
"""
Main class to use the framework, this is the main entry point to configure the framework.
The UseFramework class provides:
- Dependency injection registration via add_dependency()
- Feature and ApplicationService decorators
- Middleware registration
- Error handler configuration
- Framework execution as a callable
Example:
# Setup
framework = UseFramework("my_context")
framework.add_dependency("database", db_instance)
# Register handlers
@framework.feature(MyDTO)
class MyFeature(Feature): ...
# Execute
result = framework(MyDTO(param="value"), MyResponseDTO)
"""
log_after_execution: bool
log_app_services: bool
log_features: bool
# Decorators with better typing
feature: Callable[[Type[DataTransferObject]], DecoratorFunction]
app_service: Callable[[Type[DataTransferObject]], DecoratorFunction]
middleware_pipeline: MiddlewarePipeline
dynamic_dep_registry: Dict[str, Any]
global_error_handler: Callable[..., Any] | None
feature_error_handler: Callable[..., Any] | None
app_service_error_handler: Callable[..., Any] | None
was_initialized: bool
# Override the parent bus type to allow None during initialization
bus: FrameworkBus | None # type: ignore[override]
def __init__(
self,
bundled_context_name: str = "sincpro_framework",
log_after_execution: bool = True,
log_app_services: bool = True,
log_features: bool = True,
) -> None:
"""
Initialize the framework.
Args:
bundled_context_name: Name of the logger context
log_after_execution: Enable/disable execution logging
log_app_services: Enable/disable application service logging
log_features: Enable/disable feature logging
"""
...
# Improved overloads for framework execution
@overload
def __call__(
self, dto: DataTransferObject, return_type: Type[TypeDTOResponse]
) -> TypeDTOResponse:
"""
Execute a DTO with specified return type for better IDE support.
Args:
dto: The Data Transfer Object to execute
return_type: The expected response DTO type
Returns:
The response DTO of the specified type
"""
...
@overload
def __call__(self, dto: DataTransferObject) -> DataTransferObject | None:
"""
Execute a DTO without specifying return type.
Args:
dto: The Data Transfer Object to execute
Returns:
The response DTO or None if no response
"""
...
def build_root_bus(self) -> None:
"""Build the root bus with the dependencies provided by the user."""
...
def add_dependency(self, name: str, dep: Any) -> None:
"""
Add a dependency to the framework.
The dependency will be injected as an attribute into all Features and ApplicationServices.
Args:
name: The attribute name for the dependency
dep: The dependency instance, function, or class
Raises:
DependencyAlreadyRegistered: If the dependency name is already registered
"""
...
def add_middleware(self, middleware: Middleware) -> None:
"""
Add middleware function to the execution pipeline.
Args:
middleware: The middleware function to add
"""
...
def add_global_error_handler(self, handler: Callable[..., Any]) -> None:
"""
Add a global error handler for all framework errors.
Args:
handler: A callable that handles exceptions
"""
...
def add_feature_error_handler(self, handler: Callable[..., Any]) -> None:
"""
Add an error handler specifically for Feature errors.
Args:
handler: A callable that handles Feature exceptions
"""
...
def add_app_service_error_handler(self, handler: Callable[..., Any]) -> None:
"""
Add an error handler specifically for ApplicationService errors.
Args:
handler: A callable that handles ApplicationService exceptions
"""
...
def context(self, context_to_set: Dict[str, Any]) -> FrameworkContext:
"""
Create a context manager that applies the specified context attributes.
When used with 'with' statement, returns this UseFramework instance
with the context applied, allowing for scoped context execution.
Args:
context_to_set: Dictionary of context attributes to set
Returns:
FrameworkContext that yields this UseFramework instance when entered
Example:
with framework.context({"user_id": "123", "correlation_id": "abc"}) as app_with_context:
# app_with_context is the same UseFramework instance but with context applied
result = app_with_context(some_dto) # DTO handlers can access the context
"""
@property
def logger(self) -> LoggerProxy:
"""Get the framework logger instance."""
...