unitelabs.cdk.compose_app

Attributes

  • Name
    T
    Type
    Value

    = typing.TypeVar('T')

    Description

  • Name
    Generator
    Type
    Value

    = typing.Union[collections.abc.Generator[T, None, None], collections.abc.AsyncGenerator[T, None]]

    Description

  • Name
    AppFactory
    Type
    Value

    = collections.abc.Callable[..., typing.Union[Connector, collections.abc.Awaitable[Connector], Generator]]

    Description

Functions

  • create_logger(name : str, level : logging._Level) -> logging.Logger

    Get the app's logger and configure it if needed.

    Parameters

    • Name
      name
      Type
      str
      Default
      = __package__
      Description

    • Name
      level
      Type
      logging._Level
      Default
      = logging.INFO
      Description

    Response

    Type
    logging.Logger
    Description

  • compose_app(create_app : AppFactory, log_level : int) -> Connector

    Creates and configures a ready to use Connector.

    Parameters

    • Name
      create_app
      Type
      AppFactory
      Default
      Description

      The factory method used to create the Connector instance.

    • Name
      log_level
      Type
      int
      Default
      = logging.INFO
      Description

    Response

    Type
    Connector
    Description

    The newly created Connector.

    Exceptions

    Type
    ValueError
    Description
    If create_app does not follow the required interface.
  • signal_handler(signum : int, frame : typing.Optional[types.FrameType]) -> None

    Gracefully handle received signals.

    Parameters

    • Name
      signum
      Type
      int
      Default
      Description

      The signal number.

    • Name
      frame
      Type
      typing.Optional[types.FrameType]
      Default
      Description

      The current stack frame.

  • init_app(app_factory : AppFactory) -> Connector

    Uses the provided factory method to init a new Connector.

    Parameters

    • Name
      app_factory
      Type
      AppFactory
      Default
      Description

      The factory method to call.

    Response

    Type
    Connector
    Description

    The initialized Connector.

    Exceptions

    Type
    ValueError
    Description
    If app_factory does not follow the required interface.
  • _shutdown_yield(generator : collections.abc.AsyncGenerator[T, None]) -> None

    Execute the shutdown of a factory function by advancing the iterator after the yield and ensure the iteration ends (if not it means there is more than one yield in the function).

    Parameters

    • Name
      generator
      Type
      collections.abc.AsyncGenerator[T, None]
      Default
      Description

      The factory function to create the app.

  • _sync_to_async_gen(generator : Generator[T]) -> collections.abc.AsyncGenerator[T, None]

    Wraps any generator into an async generator.

    Parameters

    • Name
      generator
      Type
      Generator[T]
      Default
      Description

      The generator to wrap as async.

    Response

    Type
    collections.abc.AsyncGenerator[T, None]
    Description

    The async generator.

Copyright © 2024