svcs

https://svcs.hynek.me/en/stable/
Admin
svcs is a dependency container and service locator for Python that provides a central place to...
Tokens:42,417
Snippets:350
Trust Score:10
Update:1 month ago
Show doc for...
Tokens:
Raw
### get method implementation

Source: https://svcs.hynek.me/en/stable/_modules/svcs/_core.html

Implementation of the get method. It retrieves services, instantiating them if necessary and registering cleanup. Raises TypeError if an async factory is used with get. Returns a single service or a tuple of services.

```python
    def get(self, *svc_types: _ServiceType) -> object:
        """
        Get services of *svc_types*.

        Instantiate them if necessary and register their cleanup.

        Returns:
             ``svc_types[0]`` | ``tuple[*svc_types]``: If one service is
             requested, it's returned directly. If multiple are requested, a
             tuple of services is returned.
        """
        rv = []
        for svc_type in svc_types:
            cached, svc, rs = self._lookup(svc_type)
            if cached:
                rv.append(svc)
                continue

            if not isinstance(svc, MagicMock) and (
                iscoroutine(svc)
                or isinstance(svc, AbstractAsyncContextManager)
            ):
                msg = "Use `aget()` for async factories."
                raise TypeError(msg)

            if rs.enter and isinstance(svc, AbstractContextManager):
                self._on_close.append((rs, svc))
                svc = svc.__enter__()

            self._instantiated[svc_type] = (svc, rs)

            rv.append(svc)

        if len(rv) == 1:
            return rv[0]

        return rv
```

--------------------------------

### Example debug output for registrations

Source: https://svcs.hynek.me/en/stable/core-concepts.html

Sample output showing how svcs logs the registration of factories and values along with the corresponding stack trace.

```text
svcs: registered factory <built-in method now of type object at 0x103468980> for service type datetime.datetime
Stack (most recent call last):
  File "/Users/hynek/FOSS/svcs/docs/examples/debugging_with_logging.py", line 41, in <module>
    reg.register_factory(datetime, datetime.now)
  File "/Users/hynek/FOSS/svcs/src/svcs/_core.py", line 216, in register_factory
    log.debug(
svcs: registered value 'Hello World' for service type builtins.str
Stack (most recent call last):
  File "/Users/hynek/FOSS/svcs/docs/examples/debugging_with_logging.py", line 42, in <module>
    reg.register_value(str, "Hello World")
  File "/Users/hynek/FOSS/svcs/src/svcs/_core.py", line 252, in register_value
    log.debug(
```

--------------------------------

### Autowiring dependencies with svcs.autowire

Source: https://svcs.hynek.me/en/stable/autowiring.html

Demonstrates registering factories using manual dependency resolution versus the automated svcs.autowire approach. The example shows that autowired services share dependencies while remaining distinct instances.

```python
>>> from dataclasses import dataclass
>>> from typing import NewType

>>> import svcs

>>> @dataclass
... class Database: ...

>>> @dataclass
... class Cache: ...

>>> registry = svcs.Registry()
>>> registry.register_factory(Database, Database)
>>> registry.register_factory(Cache, Cache)

>>> @dataclass
... class AppServices:
...     db: Database
...     cache: Cache

>>> # BEFORE: wire the dependencies by hand.
>>> BeforeAppServices = NewType("BeforeAppServices", AppServices)
>>> registry.register_factory(
...     BeforeAppServices,
...     lambda svcs_container: AppServices(
...         db=svcs_container.get(Database),
...         cache=svcs_container.get(Cache),
...     ),
... )

>>> # AFTER: the type annotations are enough.
>>> AfterAppServices = NewType("AfterAppServices", AppServices)
>>> registry.register_factory(AfterAppServices, svcs.autowire(AppServices))

>>> with svcs.Container(registry) as container:
...     before = container.get(BeforeAppServices)
...     after = container.get(AfterAppServices)

>>> # The resulting services look the same...
>>> before
AppServices(db=Database(), cache=Cache())
>>> after
AppServices(db=Database(), cache=Cache())
>>> before == after
True
>>> # but they are different instances...
>>> before is after
False
>>> # ... although their dependencies are shared.
>>> before.db is after.db
True
>>> before.cache is after.cache
True
```

--------------------------------

### Dependency Injection via Parameters

Source: https://svcs.hynek.me/en/stable/glossary.html

Refactored version of the previous example where dependencies are passed in as parameters, making the function easier to test.

```python
def add_user(email, smtp, db):
    try:
        user = db.create_user(email)
        smtp.send_welcome_email(user)
    except DuplicateUserError:
        log.warning("Duplicate user", email=email)
```

--------------------------------

### Basic svcs.Container get() and membership check

Source: https://svcs.hynek.me/en/stable/core-concepts.html

Shows basic container usage: checking membership, retrieving a service, and confirming that repeated get() calls return the same instance. The output demonstrates that the container caches instances, which is good for DB connections but not for UUIDs.

```python
>>> container = svcs.Container(registry)

>>> uuid.UUID in container
False
>>> u = container.get(uuid.UUID)
>>> u
UUID('...')
>>> uuid.UUID in container
True
>>> # Calling get() again returns the SAME UUID instance!
>>> # Good for DB connections, bad for UUIDs.
>>> u is container.get(uuid.UUID)
True
>>> container.get(str)
'Hello World'

```

--------------------------------

### Hard-coded Dependencies in add_user

Source: https://svcs.hynek.me/en/stable/glossary.html

An example of code that is difficult to test because it constructs its own dependencies (SmtpSender and DbConnection).

```python
def add_user(email):
    smtp = SmtpSender()
    db = get_database_connection()

    try:
        user = db.create_user(email)
        smtp.send_welcome_email(user)
    except DuplicateUserError:
        log.warning("Duplicate user", email=email)
```

--------------------------------

### get(svc_types)

Source: https://svcs.hynek.me/en/stable/integrations/flask.html

Gets a service from the current container.

```APIDOC
## get(svc_types)

### Description
Retrieves a service from the container associated with flask.g.

### Method
get

### Parameters
#### Path Parameters
- **svc_types** (type or tuple) - Required - The service type(s) to retrieve.

### Returns
The requested service instance.

```

--------------------------------

### Implement a health check endpoint

Source: https://svcs.hynek.me/en/stable/integrations/aiohttp.html

This example demonstrates how to create a health check view that pings all registered external services using svcs.aiohttp.get_pings.

```python
from __future__ import annotations

from aiohttp.web import Request, Response, json_response

import svcs


async def healthy_view(request: Request) -> Response:
    """
    Ping all external services.
    """
    ok: list[str] = []
    failing: dict[str, str] = {}
    code = 200

    for svc in svcs.aiohttp.get_pings(request):
        try:
            await svc.aping()
            ok.append(svc.name)
        except Exception as e:
            failing[svc.name] = repr(e)
            code = 500

    return json_response({"ok": ok, "failing": failing}, status=code)
```

--------------------------------

### aget method implementation

Source: https://svcs.hynek.me/en/stable/_modules/svcs/_core.html

Implementation of the aget method. It is the asynchronous counterpart to get, instantiating services asynchronously if necessary. Also works with synchronous services. Since version 25.1.0, synchronous context managers are entered/exited.

```python
    async def aget(self, *svc_types: _ServiceType) -> object:
        """
        Same as :meth:`get` but instantiates asynchronously, if necessary.

        Also works with synchronous services, so in an async application, just
        use this.

        .. versionchanged:: 25.1.0
           Synchronous context managers are now entered/exited, too.
        """
        rv = []
        for svc_type in svc_types:
            cached, svc, rs = self._lookup(svc_type)
            if cached:
                rv.append(svc)
                continue
```

--------------------------------

### Defining a Starlette application with svcs

Source: https://svcs.hynek.me/en/stable/integrations/starlette.html

A simple Starlette application setup using svcs for dependency management, including a lifespan function to register a database factory.

```python
from __future__ import annotations

import os

from collections.abc import AsyncGenerator

from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

import svcs


config = {"db_url": os.environ.get("DB_URL", "sqlite:///:memory:")}


class Database:
    @classmethod
    async def connect(cls, db_url: str) -> Database:
        # ...
        return Database()

    async def get_user(self, user_id: int) -> dict[str, str]:
        return {}  # not interesting here


async def get_user(request: Request) -> JSONResponse:
    db = await svcs.starlette.aget(request, Database)

    try:
        return JSONResponse(
            {"data": await db.get_user(request.path_params["user_id"])}
        )
    except Exception as e:
        return JSONResponse({"oh no": e.args[0]})


@svcs.starlette.lifespan
async def lifespan(
    app: Starlette, registry: svcs.Registry
) -> AsyncGenerator[dict[str, object], None]:
    async def connect_to_db() -> Database:
        return await Database.connect(config["db_url"])

    registry.register_factory(Database, connect_to_db)

    yield {"your": "other stuff"}


app = Starlette(
    lifespan=lifespan,
    middleware=[Middleware(svcs.starlette.SVCSMiddleware)],
    routes=[Route("/users/{user_id}", get_user)],
)
```

--------------------------------

### Classic Service Location in Service Layer

Source: https://svcs.hynek.me/en/stable/glossary.html

An example of service location inside the service layer, which is generally discouraged due to testing and reasoning difficulties.

```python
from flask import request


def view():
    if request.form.valid():
        return handle_form_data(form.data)


def handle_form_data(data):
    svcs.flask.get(Database).do_database_stuff(data)
```

--------------------------------

### svcs.fastapi.get_registry

Source: https://svcs.hynek.me/en/stable/integrations/fastapi.html

Gets the registry that `lifespan` has attached to the app. The registry is attached when the application starts, so this only works on a running application.

```APIDOC
## get_registry

### Description
Gets the registry that `lifespan` has attached to the app. The registry is attached when the application starts, so this only works on a running application.

### Parameters
- **app** (_FastAPI_ | _TestClient_) - Required - A FastAPI application with a _svcs_-aware lifespan, or a `fastapi.testclient.TestClient` wrapping one.

### Raises
- **LookupError** - If no registry is attached to the app.

```

--------------------------------

### Service Locator Pattern with svcs_from

Source: https://svcs.hynek.me/en/stable/glossary.html

Demonstrates acquiring a database connection on-demand within a view, which is a recommended way to use svcs for dependency injection.

```python
def view(request):
    if request.form.valid():
        # Form is valid; only NOW get a DB connection
        # and pass it into your service layer.
        return handle_form_data(
            svcs_from(request).get(Database),
            form.data,
        )

    raise InvalidFormError()
```

--------------------------------

### init_app

Source: https://svcs.hynek.me/en/stable/_modules/svcs/aiohttp.html

Initializes an aiohttp application with the svcs middleware and registry.

```APIDOC
## init_app(app: web.Application, *, registry: svcs.Registry | None = None, middleware_pos: int = 0) -> web.Application

### Description
Initialize the application. Inserts the *svcs* middleware at *middleware_pos* which is 0 by default, so you can use :func:`svcs_from` and :func:`aget` in other middlewares.

### Parameters
- **app** (web.Application) - Required - The aiohttp application to initialize.
- **registry** (svcs.Registry | None) - Optional - The registry to use; defaults to a new registry.
- **middleware_pos** (int) - Optional - Position to insert the middleware; defaults to 0.

### Returns
- **web.Application** - The initialized application.
```

--------------------------------

### get method overloads

Source: https://svcs.hynek.me/en/stable/_modules/svcs/_core.html

Overloads for the get method, supporting 1 to 10 service types. The last overload returns a tuple of the requested types.

```python
    @overload
    def get(
        self,
        svc_type1: TypeForm[T1],
        svc_type2: TypeForm[T2],
        svc_type3: TypeForm[T3],
        svc_type4: TypeForm[T4],
        svc_type5: TypeForm[T5],
        svc_type6: TypeForm[T6],
        svc_type7: TypeForm[T7],
        svc_type8: TypeForm[T8],
        svc_type9: TypeForm[T9],
        svc_type10: TypeForm[T10],
        /,
    ) -> tuple[T1, T2, T3, T4, T5, T6, T7, T8, T9, T10]: ...
```

--------------------------------

### svcs_from(request)

Source: https://svcs.hynek.me/en/stable/integrations/starlette.html

Gets the current container from the request.

```APIDOC
## svcs_from(request)

### Description
Gets the current container from the request.

### Parameters
- **request** (Request) - Required - The Starlette request object.
```

--------------------------------

### init_app

Source: https://svcs.hynek.me/en/stable/_modules/svcs/flask.html

Initializes a Flask application for use with svcs, setting up the registry and teardown handler.

```APIDOC
## init_app(app: FlaskAppT, *, registry: Registry | None = None) -> FlaskAppT

### Description
Initialize *app* for *svcs*. Creates a registry for you if you don't provide one.

### Method
Function

### Parameters
- **app** (FlaskAppT) - Required - The Flask application to initialize.
- **registry** (Registry | None) - Optional - A pre-existing registry. If None, a new one is created.

### Returns
- **FlaskAppT** - The same Flask application instance.

```

--------------------------------

### get_pings()

Source: https://svcs.hynek.me/en/stable/integrations/flask.html

Gets health check pings from the current container.

```APIDOC
## get_pings()

### Description
Retrieves pings from the container, as described in svcs.Container.get_pings().

### Method
get_pings

### Returns
A list of ping results.

```

--------------------------------

### Testing Starlette views with mocked dependencies

Source: https://svcs.hynek.me/en/stable/integrations/starlette.html

Demonstrates how to use pytest and TestClient to overwrite a service in the registry after the application is ready, ensuring the mock is used during the request.

```python
from unittest.mock import Mock

import pytest

from starlette.testclient import TestClient

import svcs

from simple_starlette_app import Database, app


@pytest.fixture(name="client")
def _client():
    with TestClient(app) as client:
        yield client


def test_db_goes_boom(client):
    """
    Database errors are handled gracefully.
    """

    # IMPORTANT: Overwriting must happen AFTER the app is ready!
    db = Mock(spec_set=Database)
    db.get_user.side_effect = Exception("boom")
    svcs.starlette.get_registry(client).register_value(Database, db)

    resp = client.get("/users/42")

    assert {"oh no": "boom"} == resp.json()
```

--------------------------------

### get_abstract(*svc_types)

Source: https://svcs.hynek.me/en/stable/integrations/flask.html

Gets an abstract service from the current container (deprecated).

```APIDOC
## get_abstract(*svc_types)

### Description
Retrieves an abstract service from the container on flask.g. Deprecated since version 26.1.0.

### Method
get_abstract

### Parameters
#### Path Parameters
- **svc_types** (type) - Required - The service type(s) to retrieve.

### Returns
The requested abstract service.

```

--------------------------------

### Composition Root using svcs

Source: https://svcs.hynek.me/en/stable/glossary.html

Shows the recommended pattern where the view acts as a composition root to acquire services and inject them into the service layer.

```python
def view(request):
    """
    View and composition root.
    """
    return do_something(svcs_from(request).get(Database))


def do_something(db):
    """
    Service layer.
    """
    db.do_database_stuff()
```

--------------------------------

### svcs.aiohttp.aget

Source: https://svcs.hynek.me/en/stable/integrations/aiohttp.html

Asynchronously gets a service from the container associated with the request, equivalent to `svcs.Container.aget()`.

```APIDOC
## svcs.aiohttp.aget

### Description
Asynchronously gets a service from the container associated with the request, equivalent to `svcs.Container.aget()`.

### Method
Async function call

### Signature
`async svcs.aiohttp.aget(request: aiohttp.web.Request, svc_type1: type, ...)`

### Parameters
- **request** (aiohttp.web.Request) - Required - The AIOHTTP request object.
- **svc_type1** (type) - Required - The service type to retrieve.
- **...** - Additional service types to retrieve.

### Returns
- The requested service instance(s).

### Usage Example
python
service = await svcs.aiohttp.aget(request, Database)

```

--------------------------------

### Registry context manager with on_registry_close

Source: https://svcs.hynek.me/en/stable/_modules/svcs/_core.html

Demonstrates using the Registry as a context manager to execute callbacks upon registry exit.

```python
>>> import svcs
>>> with svcs.Registry() as reg:
...     reg.register_value(
...         int, 42,
...         on_registry_close=lambda: print("closed!")
...     )
closed!
```

--------------------------------

### Initialize Starlette with svcs

Source: https://svcs.hynek.me/en/stable/integrations/starlette.html

Wrap the lifespan with svcs.starlette.lifespan and include SVCSMiddleware in the Starlette application configuration.

```python
from starlette.applications import Starlette
from starlette.middleware import Middleware

import svcs


@svcs.starlette.lifespan
async def lifespan(app: Starlette, registry: svcs.Registry):
    registry.register_factory(Database, Database.connect)

    yield {"your": "other stuff"}

    # Registry is closed automatically when the app is done.


app = Starlette(
    lifespan=lifespan,
    middleware=[Middleware(svcs.starlette.SVCSMiddleware)],
    routes=[...],
)
```

--------------------------------

### Initialize svcs in Flask application factory with SQLAlchemy

Source: https://svcs.hynek.me/en/stable/integrations/flask.html

Sets up the svcs registry by wrapping the Flask app with svcs.flask.init_app(). Registers a factory for SQLAlchemy Connection with cleanup via on_registry_close and a ping function, another repository factory using svcs.flask.get() inside the factory, and an atexit handler to close the registry. Use this in an application factory that creates the Flask app.

```python
import atexit

from flask import Flask
from sqlalchemy import Connection, create_engine
from sqlalchemy.sql import text

import svcs


def create_app(config_filename):
    app = Flask(__name__)

    ...

    ##########################################################################
    # Set up the registry using Flask integration.
    app = svcs.flask.init_app(app)

    # Now, register a factory that calls `engine.connect()` if you ask for a
    # `Connection`. Since we use yield inside of a context manager, the
    # connection gets cleaned up when the container is closed.
    # If you ask for a ping, it will run `SELECT 1` on a new connection and
    # clean up the connection behind itself.
    engine = create_engine("postgresql://localhost")

    def connection_factory():
        with engine.connect() as conn:
            yield conn

    ping = text("SELECT 1")
    svcs.flask.register_factory(
        # The app argument makes it good for custom init_app() functions.
        app,
        Connection,
        connection_factory,
        ping=lambda conn: conn.execute(ping).close(),
        on_registry_close=engine.dispose,
    )

    # You also use svcs WITHIN factories:
    svcs.flask.register_factory(
        app,  # <---
        AbstractRepository,
        # No cleanup, so we just return an object using a lambda
        lambda: Repository.from_connection(svcs.flask.get(Connection)),
    )

    @atexit.register
    def cleanup() -> None:
        """
        Clean up all pools when the application shuts down.
        """
        log.info("app.cleanup.start")
        svcs.flask.close_registry(app)  # calls engine.dispose()
        log.info("app.cleanup.done")

    ##########################################################################

    ...

    return app

```

--------------------------------

### Container context manager usage

Source: https://svcs.hynek.me/en/stable/_modules/svcs/_core.html

Demonstrates using the Container as a context manager to automatically trigger cleanup routines for registered services.

```python
>>> reg = svcs.Registry()
>>> def factory() -> str:
...     yield "Hello World"
...     print("Cleaned up!")
>>> reg.register_factory(str, factory)

>>> with svcs.Container(reg) as con:
...     _ = con.get(str)
Cleaned up!
```

--------------------------------

### aautowire(fn_or_cls)

Source: https://svcs.hynek.me/en/stable/autowiring.html

Like `autowire()`, but dependencies are resolved with `svcs.Container.aget()`, the returned factory is async, and the result of the callable is awaited if it's awaitable. It also works with synchronous callables and services, so in an async application, just use this.

```APIDOC
## aautowire(fn_or_cls)

### Description
Like `autowire()`, but dependencies are resolved with `svcs.Container.aget()`, the returned factory is async, and the result of the callable is awaited if it's awaitable. It also works with synchronous callables and services, so in an async application, just use this.

### Parameters
- **fn_or_cls** (_Callable_ _[__[__...__]__,___T_ _]_) - Required - A callable (function or class) whose parameters will be autowired based on their type annotations.

### Returns
A factory that takes a container and returns an awaitable result of calling the callable with the resolved dependencies.

### Return Type
_Callable_[[_Container_], _Awaitable_[__T_]]

### Notes
- Added in version 26.1.0.
```

--------------------------------

### Autowire a function with svcs.autowire

Source: https://svcs.hynek.me/en/stable/autowiring.html

Use @autowire to decorate a factory function. The function's parameters are resolved from the container based on type annotations; defaults are used if a service is not found. The example registers a value and the factory, then retrieves the result from the container.

```python
from svcs import Container, Registry, autowire


class Service:
    def __init__(self, name: str) -> None:
        self.name = name


@autowire
def build_label(svc: Service, suffix: int = 42) -> str:
    return f"{svc.name}{suffix}"


registry = Registry()
registry.register_value(Service, Service("api"))
registry.register_factory(str, build_label)

with Container(registry) as container:
    result = container.get(str)
    print(result)  # Output: "api42"
```

--------------------------------

### Register factories and values with svcs.Registry

Source: https://svcs.hynek.me/en/stable/core-concepts.html

Shows how to create a registry and register factory callables and values. The registry stores factories and values, and membership checks work with the registered types. Types must be hashable.

```python
>>> import svcs
>>> import uuid

>>> registry = svcs.Registry()

>>> registry.register_factory(uuid.UUID, uuid.uuid4)
>>> registry.register_value(str, "Hello World")
>>> uuid.UUID in registry
True
>>> str in registry
True
>>> int in registry
False
```

--------------------------------

### get_registry(app=None)

Source: https://svcs.hynek.me/en/stable/integrations/flask.html

Retrieves the registry from the app or the current app context.

```APIDOC
## get_registry(app=None)

### Description
Gets the registry from the given app or from flask.current_app if app is None.

### Method
get_registry

### Parameters
#### Query Parameters
- **app** (Flask | None) - Optional - The Flask application; if None, flask.current_app is used.

### Returns
The registry associated with the app.

```

--------------------------------

### Service layer function with injected services

Source: https://svcs.hynek.me/en/stable/glossary.html

Example of a service layer function that uses injected services (unit_of_work and mail_q) to coordinate database operations and email sending. The domain model is a pure function that enforces business rules without using services.

```python
def add_user_to_org(unit_of_work, mail_q, org_id, user_id):
    """
    Service Layer entry point.
    """
    with unit_of_work:
        org = unit_of_work.organizations.get(org_id)
        user = unit_of_work.users.get(user_id)

        domain_model.check_if_can_add_user_to_org(org, user)

        unit_of_work.organizations.add_user(org_id, user_id)
        mail_q.send_welcome_mail(user.email, user.name, org.name)

        unit_of_work.commit()

```

--------------------------------

### Recursive service acquisition with svcs_container factory

Source: https://svcs.hynek.me/en/stable/core-concepts.html

Demonstrates recursive service acquisition: a factory that takes a first argument named svcs_container (or annotated as svcs.Container) receives the current container as a positional argument, allowing it to fetch other services. The example registers a value and a factory, then retrieves the string service.

```python
>>> container = svcs.Container(registry)

# Let's make the UUID predictable for our test!
>>> registry.register_value(uuid.UUID, uuid.UUID('639c0a5c-8d93-4a67-8341-fe43367308a5'))

>>> def factory(svcs_container) -> str:
...     return svcs_container.get(uuid.UUID).hex  # get the UUID, then work on it

>>> registry.register_factory(str, factory)

>>> container.get(str)
'639c0a5c8d934a678341fe43367308a5'

```

--------------------------------

### Register a cleanup callback with on_registry_close

Source: https://svcs.hynek.me/en/stable/core-concepts.html

Shows how to register a callback that is called when the registry is closed. If the callback fails, it is logged at warning level and ignored. This can be used to free resources like database connection pools.

```python
engine = create_engine(url)

registry.register_factory(
    Connection, engine.connect, on_registry_close=engine.dispose
)
```

--------------------------------

### register_factory(app, svc_type, factory, *, enter=True, ping=None, on_registry_close=None)

Source: https://svcs.hynek.me/en/stable/integrations/flask.html

Registers a factory for a service type on the app's registry.

```APIDOC
## register_factory(app, svc_type, factory, *, enter=True, ping=None, on_registry_close=None)

### Description
Registers a factory for a service type on the app's registry, similar to svcs.Registry.register_factory().

### Method
register_factory

### Parameters
#### Path Parameters
- **app** (Flask) - Required - The Flask application.
- **svc_type** (type) - Required - The service type to register.
- **factory** (callable) - Required - The factory that creates the service.

#### Keyword Arguments
- **enter** (bool) - Optional - Whether to enter the service context (default True).
- **ping** (callable, optional) - Optional - A ping function for health checks.
- **on_registry_close** (callable, optional) - Optional - Callback when the registry is closed.

### Returns
None

```

--------------------------------

### Register multiple factories for the same type using subclassing, NewType, and Annotated

Source: https://svcs.hynek.me/en/stable/core-concepts.html

Demonstrates three approaches to register multiple factories for the same type: subclassing, typing.NewType, and typing.Annotated. Note that NewType and Annotated do not work with Pyright, and the metadata content is not important as long as the type is hashable.

```python
from typing import Annotated, NewType

from sqlalchemy import Connection, create_engine

# Create the two connection engines
primary_engine = create_engine(primary_url)
secondary_engine = create_engine(secondary_url)
tertiary_engine = create_engine(tertiary_url)


# Clunky, but works universally:
class PrimaryConnection(Connection):
    pass


# Works with Mypy, ty, and Pyrefly, but NOT with Pyright:
SecondaryConnection = NewType("SecondaryConnection", Connection)

# Works with Mypy, ty, and Pyrefly, but NOT with Pyright:
TertiaryConnection = Annotated[Connection, "tertiary"]

# Register the factories to the aliases
registry.register_factory(PrimaryConnection, primary_engine.connect)
registry.register_factory(SecondaryConnection, secondary_engine.connect)
registry.register_factory(TertiaryConnection, tertiary_engine.connect)
```

--------------------------------

### Acquire services using svcs_from

Source: https://svcs.hynek.me/en/stable/integrations/starlette.html

Use svcs_from to extract the container from the request and retrieve services.

```python
from svcs.starlette import svcs_from


async def view(request):
    db = await svcs_from(request).aget(Database)
```

--------------------------------

### RegisteredService class definition

Source: https://svcs.hynek.me/en/stable/_modules/svcs/_core.html

Defines a recipe for creating a service, including factory configuration and context management settings.

```python
@attrs.frozen
class RegisteredService:
    """
    A recipe for creating a service.

    .. warning::

        Strictly read-only.

    Attributes:
        svc_type: The type under which the type has been registered.

        factory: Callable that creates the service.

        takes_container:
            Whether the factory takes a container as its first argument.

        enter: Whether context managers returned by the factory are entered.

        ping: See :ref:`health`.

        suppress_context_exit:
            Whether to suppress errors raised in the container context when
            exiting registered factories.
    """

    svc_type: _ServiceType
    factory: Callable = attrs.field(hash=False)
    takes_container: bool
    enter: bool
    ping: Callable | None = attrs.field(hash=False)
    suppress_context_exit: bool

    @property
    def name(self) -> str:
        return _full_name(self.svc_type)

    def __repr__(self) -> str:
        return (
            f"<RegisteredService(svc_type="
            f"{self.name}, "
            f"factory={self.factory}, "
            f"takes_container={self.takes_container}, "
            f"enter={self.enter}, "
            f"has_ping={self.ping is not None}, "
            f"suppress_context_exit={self.suppress_context_exit}"
            ")>"
        )

    def close(
        self,
        cm: AbstractContextManager,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        """
        Close the context manager *cm* with the given exception information.
        """
        if self.suppress_context_exit:
            cm.__exit__(None, None, None)
        else:
            cm.__exit__(exc_type, exc_val, exc_tb)

    async def aclose(
        self,
        cm: AbstractAsyncContextManager,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        """
        Close the async context manager *cm* with the given exception information.
        """
        if self.suppress_context_exit:
            await cm.__aexit__(None, None, None)
        else:
            await cm.__aexit__(exc_type, exc_val, exc_tb)
```

--------------------------------

### Autowire a class with Registry

Source: https://svcs.hynek.me/en/stable/autowiring.html

Use autowire as a wrapper during registration to automatically resolve dependencies for a class constructor.

```python
from svcs import Container, Registry, autowire


class Service:
    def __init__(self, name: str) -> None:
        self.name = name


class Handler:
    def __init__(self, svc: Service, prefix: str = "svc:") -> None:
        self.svc = svc
        self.prefix = prefix


registry = Registry()
registry.register_value(Service, Service("api"))
registry.register_factory(Handler, autowire(Handler))

with Container(registry) as container:
    handler = container.get(Handler)
    print(handler.svc.name, handler.prefix)  # Output: "api svc:"
```

--------------------------------

### svcs.aiohttp.init_app

Source: https://svcs.hynek.me/en/stable/integrations/aiohttp.html

Initializes the svcs middleware on an AIOHTTP application, inserting it at the specified middleware position (default 0) so that svcs_from() and aget() can be used in other middlewares.

```APIDOC
## svcs.aiohttp.init_app

### Description
Initializes the application by inserting the svcs middleware at the given position (default 0) so that `svcs_from()` and `aget()` can be used in other middlewares.

### Method
Function call

### Signature
`svcs.aiohttp.init_app(app, *, registry=None, middleware_pos=0)`

### Parameters
- **app** (aiohttp.web.Application) - Required - The AIOHTTP application to initialize.
- **registry** (svcs.Registry, optional) - Optional - The registry to use; defaults to a new registry.
- **middleware_pos** (int, optional) - Optional - Position to insert the middleware; defaults to 0.

### Usage Example
python
from aiohttp import web
import svcs

app = web.Application()
svcs.aiohttp.init_app(app)

```

--------------------------------

### Retrieve services with svcs in AIOHTTP, FastAPI, Flask, and Starlette

Source: https://svcs.hynek.me/en/stable/why.html

Shows how to retrieve services in each supported web framework. In AIOHTTP and Starlette, pass the request to aget; in FastAPI, use the DepContainer dependency; in Flask, call svcs.flask.get directly. The type checker infers Database, WebAPIClient, and Cache from the requested types.

```python
import svcs


async def view(request):
    db, api, cache = await svcs.aiohttp.aget(
        request, Database, WebAPIClient, Cache
    )

    ...

```

```python
import svcs


@app.get("/")
async def view(services: svcs.fastapi.DepContainer):
    db, api, cache = await services.aget(Database, WebAPIClient, Cache)

    ...

```

```python
import svcs


@app.route("/")
def view():
    db, api, cache = svcs.flask.get(Database, WebAPIClient, Cache)

    ...

```

```python
import svcs


async def view(request):
    db, api, cache = await svcs.starlette.aget(
        request, Database, WebAPIClient, Cache
    )

    ...

```

--------------------------------

### Re-export svcs.flask helpers and use them in app factory and views

Source: https://svcs.hynek.me/en/stable/integrations/flask.html

Create a module that re-exports Flask helpers from svcs.flask to simplify view code. Then use that module in an application factory to initialize the registry and register a factory, and in a view to retrieve the service. The first block defines the re-export module, the second shows registration in an app factory, and the third shows service acquisition in a view.

```python
from svcs.flask import (
    close_registry,
    get,
    get_pings,
    init_app,
    overwrite_factory,
    overwrite_value,
    register_factory,
    register_value,
    svcs_from,
)


__all__ = [
    "close_registry",
    "get_pings",
    "get",
    "init_app",
    "overwrite_factory",
    "overwrite_value",
    "register_factory",
    "register_value",
    "svcs_from",
]

```

```python
from your_app import services


def init_app(app):
    app = services.init_app(app)
    services.register_factory(app, Connection, ...)

    return app

```

```python
from your_app import services


@app.route("/")
def index():
    conn = services.get(Connection)

```

--------------------------------

### Acquire services in AIOHTTP

Source: https://svcs.hynek.me/en/stable/index.html

Use svcs.aiohttp.aget to acquire multiple services within an AIOHTTP request handler.

```python
import svcs


async def view(request):
    db, api, cache = await svcs.aiohttp.aget(
        request, Database, WebAPIClient, Cache
    )

    ...
```

--------------------------------

### svcs._autowire module source: autowire() and helpers

Source: https://svcs.hynek.me/en/stable/_modules/svcs/_autowire.html

Complete source of the svcs._autowire module. It defines the autowire() decorator and internal helpers (_lazy_signature, _reject_bare_generator, _wireable_params, _default_or_raise) that resolve a callable's annotated parameters from a svcs.Container. Use this module when you need to build factories that automatically inject services based on type hints, with support for defaults, dataclasses.InitVar, and Container parameters. Bare generator factories are rejected with a TypeError; decorate them with contextlib.contextmanager or contextlib.asynccontextmanager instead.

```python
# SPDX-FileCopyrightText: 2023 Hynek Schlawack <hs@ox.cx>
#
# SPDX-License-Identifier: MIT

from __future__ import annotations

import dataclasses
import inspect
import sys

from collections.abc import Awaitable, Callable, Iterator
from contextlib import AbstractAsyncContextManager, AbstractContextManager
from typing import Any, ForwardRef, TypeVar, overload

from ._core import Container, _robust_signature
from .exceptions import ServiceNotFoundError


if sys.version_info >= (3, 14):
    from typing import evaluate_forward_ref
else:
    _A = TypeVar("_A")

    def evaluate_forward_ref(annotation: _A) -> _A:
        return annotation


_T = TypeVar("_T")


def _lazy_signature(
    fn_or_cls: Callable[..., Any],
) -> Callable[[], inspect.Signature]:
    """
    Return a callable that resolves and caches the signature of *fn_or_cls* on
    first use.

    This allows autowire to work with (eventually resolvable) forward
    references.
    """
    cache: inspect.Signature | None = None

    def get_signature() -> inspect.Signature:
        nonlocal cache

        if not cache:
            sig = _robust_signature(fn_or_cls)
            if sig is None:
                msg = f"Cannot determine the signature of {fn_or_cls!r}."
                raise TypeError(msg)

            cache = sig

        return cache

    return get_signature


def _reject_bare_generator(fn_or_cls: Callable[..., Any]) -> None:
    """
    Raise TypeError if *fn_or_cls* is a bare (async) generator function.

    The autowire wrapper is a plain function, so the registry's automatic
    context-manager conversion of generator factories never kicks in and
    their cleanup would be silently lost.
    """
    if inspect.isgeneratorfunction(fn_or_cls):
        msg = (
            f"Cannot autowire the generator function {fn_or_cls!r}: its "
            f"cleanup would be lost. Decorate it with "
            f"contextlib.contextmanager instead."
        )
        raise TypeError(msg)

    if inspect.isasyncgenfunction(fn_or_cls):
        msg = (
            f"Cannot autowire the async generator function {fn_or_cls!r}: "
            f"its cleanup would be lost. Decorate it with "
            f"contextlib.asynccontextmanager instead."
        )
        raise TypeError(msg)


def _wireable_params(
    sig: inspect.Signature,
) -> Iterator[tuple[str, inspect.Parameter, Any]]:
    """
    Yield the ``(name, parameter, annotation)`` triples of *sig* that
    autowire should resolve from a container.
    """
    for name, param in sig.parameters.items():
        # skip variadic parameters (``*args`` and ``**kwargs``)
        if param.kind in (param.VAR_POSITIONAL, param.VAR_KEYWORD):
            continue

        annotation = param.annotation
        if isinstance(annotation, dataclasses.InitVar):
            annotation = annotation.type

        if annotation is param.empty and param.default is param.empty:
            msg = (
                f"Cannot autowire parameter {name!r}: it has no type "
                f"annotation and no default."
            )
            raise TypeError(msg)

        if isinstance(annotation, ForwardRef):
            annotation = evaluate_forward_ref(annotation)

        yield name, param, annotation


def _default_or_raise(
    param: inspect.Parameter, annotation: Any, exc: ServiceNotFoundError
) -> Any:
    """
    Return *param*'s default when its own service is missing, else re-raise.
    """
    # Only fall back for a miss of *annotation* itself. A miss for a different
    # type means a registered factory failed to resolve its own dependency,
    # which must not be masked by the default.
    if param.default is param.empty or exc.args[0] is not annotation:
        raise exc

    return param.default




[docs]
def autowire(fn_or_cls: Callable[..., _T]) -> Callable[[Container], _T]:
    """
    Return a factory that resolves the dependencies of *fn_or_cls* from
    a container, based on its type annotations.

    The returned factory takes a :class:`svcs.Container`, resolves every
    annotated parameter of *fn_or_cls* from it, and calls *fn_or_cls* with the
    resolved services.

    If a service is not found in the container and the parameter has a default
    value, the default is used instead of raising an error.

    Variadic parameters (``*args`` and ``**kwargs``) are ignored and
    :class:`dataclasses.InitVar` annotations are unwrapped.

    A parameter annotated as :class:`svcs.Container` receives the current
    container itself.

    Factories that return context managers are entered and cleaned up as usual.
    Bare generator factories are rejected, because their cleanup would be lost.
    Decorate them with :func:`~contextlib.contextmanager` instead.

    Args:
        fn_or_cls:
            A callable (function or class) whose parameters will be
            autowired based on their type annotations.

```
Tokens:
Raw