autowire package

Submodules

autowire.base_container module

autowire.base

Base definitions of autowire.

class autowire.base_container.BaseContainer(parent: Optional[autowire.base_container.BaseContainer] = None)

Bases: abc.ABC

Dependency injection container base class.

contextual(resource: autowire.resource.Resource[~R][R], *arg_resources, **kwarg_resources) → Callable[[C], C]

Provide resource’s implementation with context manager

arg_resources and kwarg_resources will be used for dependency injection.

db_connection = Resource("db_connection", __name__)
db_transaction = Resource("transaction", __name__)

container = Container()

@container.contextual(db_transaction, db_connection)
@contextlib.contextmanager
def begin_trasaction(db_connection: Connection):
    tx = db_connection.begin()
    try:
        yield tx
    except Exception:
        tx.rollback()
        raise
    finally:
        tx.commit()

find_implementation(resource: autowire.base_resource.BaseResource[~R][R]) → autowire.implementation.Implementation[~R][R]

plain(resource: autowire.resource.Resource[~R][R], *arg_resources, **kwarg_resources) → Callable[[F], F]

Provide resource’s implementation with plain function

arg_resources and kwarg_resources will be used for dependency injection.

config = Resource("config", __name__)
connection_pool = Resource("connection_pool", __name__)
db_connection = Resource("db_connection", __name__)

container = Container()

@container.plain(db_connection, connection_pool, config=config)
def get_db_connection(connection_pool: Pool, *, config: dict) -> Connection:
    # ...

provide(resource: autowire.resource.Resource[~R][R], implementation: autowire.implementation.Implementation[~R][R])

Provide an implementation for resource.

provide_constant(resource: autowire.resource.Resource[~R][R], constant: R)

Provide resource’s implementation with constant implementation that holds given constant as a value.

global_config = Resource("global_config", __name__)

container = Container()

container.provide_constant(global_config, {"DB_TIMEOUT": 30})

autowire.base_resource module

class autowire.base_resource.BaseResource(name: str, namespace: str)

Bases: abc.ABC, typing.Generic

Decalarative resource definition.

canonical_name

Canonical name of resource.

It’s <namespace>.<name>

autowire.container module

class autowire.container.Container(parent: Optional[autowire.base_container.BaseContainer] = None)

Bases: autowire.base_container.BaseContainer

Dependency injection container

context(preload: Sequence[autowire.base_resource.BaseResource] = ()) → Iterator[autowire.context.Context]

Get a DI context from this container.

with container.context() as context:
    value = context.resolve(resource)
    ...

Parameters:preload – resources to be preloaded on this context.

autowire.context module

autowire.base

Base definitions of autowire.

class autowire.context.Context(container: autowire.base_container.BaseContainer, parent: Optional[autowire.context.Context])

Bases: autowire.provider.ResourceProvider, contextlib.AbstractContextManager, typing.Generic

Resource management context base class

child(preload: Sequence[autowire.base_resource.BaseResource] = ()) → Iterator[autowire.context.Context]

Create a child context

with context.child() as child:
    value = child.resolve(resource)
    ...

Parameters:preload – resources to be preloaded

drain()

Drain all resources resolved by this context.

resolve(resource: autowire.base_resource.BaseResource[~R][R]) → R

Resolve resource in this context.

exception autowire.context.NotPooled

Bases: Exception

Internal class for fiding pooled resource

autowire.exc module

autowire.exc

Exception definitions.

exception autowire.exc.ResourceNotProvidedError

Bases: RuntimeError

Error for not provided resource.

autowire.implementation module

class autowire.implementation.ConstantImplementation(value: R)

Bases: autowire.implementation.ContextManagerImplementation

Use constant as an implementation

class autowire.implementation.ContextManagerImplementation(manager_generator: Callable[[...], AbstractContextManager[R]], arg_resources: Tuple[autowire.base_resource.BaseResource[typing.Any][Any], ...], kwarg_resources: Dict[str, autowire.base_resource.BaseResource[typing.Any][Any]])

Bases: autowire.implementation.Implementation

Use context manager as an implementation

reify(resource: autowire.base_resource.BaseResource[~R][R], provider: autowire.provider.ResourceProvider)

Reify the resource

class autowire.implementation.Implementation

Bases: typing.Generic

Declarative base type for implementation

reify(resource: autowire.base_resource.BaseResource[~R][R], provider: autowire.provider.ResourceProvider) → AbstractContextManager[R]

Reify the resource

class autowire.implementation.PlainFunctionImplementation(fn: Callable[[...], R], arg_resources: Tuple[autowire.base_resource.BaseResource[typing.Any][Any], ...], kwarg_resources: Dict[str, autowire.base_resource.BaseResource[typing.Any][Any]])

Bases: autowire.implementation.ContextManagerImplementation

Use plain function as an implementation

autowire.provider module

class autowire.provider.ResourceProvider

Bases: abc.ABC

Declarative base class for resource provider

resolve(resource: autowire.base_resource.BaseResource[~R][R]) → R

autowire.resource module

class autowire.resource.Resource(name: str, namespace: str)

Bases: autowire.base_resource.BaseResource

Decalarative resource definition.

Namespace generally be the module’s name.

name cannot include any dot(.) characters.

>>> resource = BaseResource('name', __name__)

contextual(*arg_resources, **kwarg_resources) → Callable[[C], C]

Set the default implementation with context manager

arg_resources and kwarg_resources will be used for dependency injection.

db_connection = Resource("db_connection", __name__)
db_transaction = Resource("transaction", __name__)

@db_transaction.contextual(db_connection)
@contextlib.contextmanager
def begin_trasaction(db_connection: Connection):
    tx = db_connection.begin()
    try:
        yield tx
    except Exception:
        tx.rollback()
        raise
    finally:
        tx.commit()

plain(*arg_resources, **kwarg_resources) → Callable[[F], F]

Set the default implementation with plain function

arg_resources and kwarg_resources will be used for dependency injection.

config = Resource("config", __name__)
connection_pool = Resource("connection_pool", __name__)
db_connection = Resource("db_connection", __name__)

@db_connection.plain(connection_pool, config=config)
def get_db_connection(connection_pool: Pool, *, config: dict) -> Connection:
    # ...

set_constant(constant: R)

Set the default implementation with constant implementation that holds given constant as a value.

global_config = Resource("global_config", __name__)

global_config.set_constant({"DB_TIMEOUT": 30})

Module contents

class autowire.Context(container: autowire.base_container.BaseContainer, parent: Optional[autowire.context.Context])

Bases: autowire.provider.ResourceProvider, contextlib.AbstractContextManager, typing.Generic

Resource management context base class

child(preload: Sequence[autowire.base_resource.BaseResource] = ()) → Iterator[autowire.context.Context]

Create a child context

with context.child() as child:
    value = child.resolve(resource)
    ...

Parameters:preload – resources to be preloaded

drain()

Drain all resources resolved by this context.

resolve(resource: autowire.base_resource.BaseResource[~R][R]) → R

Resolve resource in this context.

class autowire.Container(parent: Optional[autowire.base_container.BaseContainer] = None)

Bases: autowire.base_container.BaseContainer

Dependency injection container

context(preload: Sequence[autowire.base_resource.BaseResource] = ()) → Iterator[autowire.context.Context]

Get a DI context from this container.

with container.context() as context:
    value = context.resolve(resource)
    ...

Parameters:preload – resources to be preloaded on this context.

class autowire.Resource(name: str, namespace: str)

Bases: autowire.base_resource.BaseResource

Decalarative resource definition.

Namespace generally be the module’s name.

name cannot include any dot(.) characters.

>>> resource = BaseResource('name', __name__)

contextual(*arg_resources, **kwarg_resources) → Callable[[C], C]

Set the default implementation with context manager

arg_resources and kwarg_resources will be used for dependency injection.

db_connection = Resource("db_connection", __name__)
db_transaction = Resource("transaction", __name__)

@db_transaction.contextual(db_connection)
@contextlib.contextmanager
def begin_trasaction(db_connection: Connection):
    tx = db_connection.begin()
    try:
        yield tx
    except Exception:
        tx.rollback()
        raise
    finally:
        tx.commit()

plain(*arg_resources, **kwarg_resources) → Callable[[F], F]

Set the default implementation with plain function

arg_resources and kwarg_resources will be used for dependency injection.

config = Resource("config", __name__)
connection_pool = Resource("connection_pool", __name__)
db_connection = Resource("db_connection", __name__)

@db_connection.plain(connection_pool, config=config)
def get_db_connection(connection_pool: Pool, *, config: dict) -> Connection:
    # ...

set_constant(constant: R)

Set the default implementation with constant implementation that holds given constant as a value.

global_config = Resource("global_config", __name__)

global_config.set_constant({"DB_TIMEOUT": 30})

exception autowire.ResourceNotProvidedError

Bases: RuntimeError

Error for not provided resource.