### Flask 'get_X' Pattern Example

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/flask.md

Demonstrates the common Flask pattern for managing dependencies using `flask.g` and `teardown_appcontext`. This pattern involves creating a function to get a resource, storing it in `g`, and registering a teardown handler to clean it up.

```python
from flask import g

def get_db():
    if 'db' not in g:
        g.db = connect_to_database()

    return g.db

@app.teardown_appcontext
def teardown_db(exception):
    db = g.pop('db', None)

    if db is not None:
        db.close()
```

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

### Health Check Endpoint Example (AIOHTTP)

Source: https://github.com/hynek/svcs/blob/main/docs/core-concepts.md

Example of a health check endpoint implementation using AIOHTTP. This snippet demonstrates how to integrate with the `svcs` library for service health monitoring.

```python
from aiohttp import web
import svcs

async def health_check(request):
    container: svcs.Container = request.app["svcs_container"]
    pings = await container.aping(svcs.ServicePing)
    return web.json_response({
        "services": [
            {
                "name": ping.name,
                "ok": True,
                "error": None
            }
            for ping in pings
        ]
    })


def setup_health_check(app: web.Application):
    app.router.add_get("/health", health_check)
```

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

### Simple FastAPI Application for Testing

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/fastapi.md

A basic FastAPI application setup with a lifespan to demonstrate how svcs manages service registries for testing purposes.

```python
from fastapi import FastAPI, Depends
from typing import Annotated

import svcs


class Database:
    async def connect(self):
        pass


class UserService:
    def __init__(self, db: Database):
        self.db = db

    def get_user(self, user_id: int):
        return {"id": user_id, "name": "Test User"}


@svcs.fastapi.lifespan
async def lifespan(app: FastAPI, registry: svcs.Registry):
    registry.register_factory(Database, Database.connect)
    registry.register_singleton(UserService)
    yield


app = FastAPI(lifespan=lifespan)


@app.get("/users/{user_id}")
def get_user(
    services: Annotated[svcs.Container, Depends(svcs.fastapi.container)],
    user_id: int,
):
    user_service = services.get(UserService)
    return user_service.get_user(user_id)
```

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

### Health Check Endpoint Example (Starlette)

Source: https://github.com/hynek/svcs/blob/main/docs/core-concepts.md

Example of a health check endpoint implementation using Starlette. This snippet demonstrates using `svcs.Container.get_pings()` for synchronous health checks.

```python
from starlette.requests import Request
from starlette.responses import JSONResponse
import svcs

def health_check(request: Request):
    container: svcs.Container = request.state.svcs_container
    pings = container.get_pings()
    return JSONResponse({
        "services": [
            {
                "name": ping.name,
                "ok": True,
                "error": None
            }
            for ping in pings
        ]
    })
```

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

### Health Check Endpoint Example (Pyramid)

Source: https://github.com/hynek/svcs/blob/main/docs/core-concepts.md

Example of a health check endpoint implementation using Pyramid. This snippet shows how to access the container from the request and use `svcs.ServicePing` for health checks.

```python
from pyramid.view import view_config
import svcs

@view_config(route_name='health', renderer='json')
def health_check(request):
    container: svcs.Container = request.registry.svcs_container
    pings = container.get_pings()
    return {
        "services": [
            {
                "name": ping.name,
                "ok": True,
                "error": None
            }
            for ping in pings
        ]
    }
```

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

### Health Check Endpoint Example (FastAPI)

Source: https://github.com/hynek/svcs/blob/main/docs/core-concepts.md

Example of a health check endpoint implementation using FastAPI. This snippet shows how to use `svcs.Container.aping` for asynchronous health checks.

```python
from fastapi import FastAPI
import svcs

def health_check(container: svcs.Container = Depends(svcs.get_container)):
    pings = container.get_pings()
    return {
        "services": [
            {
                "name": ping.name,
                "ok": True,
                "error": None
            }
            for ping in pings
        ]
    }

app = FastAPI()
app.add_api_route("/health", health_check)
```

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

### Acquire Service using svcs_from

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/starlette.md

Use svcs_from to get a service from the request context within a Starlette view.

```python
from svcs.starlette import svcs_from

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

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

### Application Life Cycle

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/flask.md

Functions for initializing, getting, and closing the service registry within a Flask application context.

```APIDOC
## Application Life Cycle

### `init_app(app)`

Initializes the service registry for a Flask application.

### `get_registry()`

Retrieves the service registry associated with the current Flask application.

### `close_registry()`

Closes the service registry for the current Flask application.

### `registry`

A `werkzeug.local.LocalProxy` that transparently calls `get_registry()` on `flask.current_app`. Available since version 23.21.0.
```

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

### Mocking Database Connection in Flask Tests

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/flask.md

Illustrates how to replace a Flask dependency with a mock object for testing purposes using `appcontext_pushed`. This example shows a custom context manager to set `g.db` and a test client to verify the behavior when the dependency is unavailable.

```python
from contextlib import contextmanager
from flask import appcontext_pushed

@contextmanager
def db_set(app, db):
    def handler(sender, **kwargs):
        g.db = db  # ← setting g.db here prevents get_db setting it itself
    with appcontext_pushed.connected_to(handler, app):
        yield

class Boom:
    def __getattr__(self, name):
        """Just raise an exception when you try to use it."""
        raise RuntimeError("Boom!")

def test_broken_db(app):
    with db_set(app, Boom()):
        c = app.test_client()
        resp = c.get('/some-url')

        assert 500 == resp.status_code
```

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

### Safer Dependency Flow: Local to Global

Source: https://github.com/hynek/svcs/blob/main/docs/core-concepts.md

It is safer to have local registries depend on global ones. This example shows registering a global connection factory and then a local factory that uses it, ensuring proper dependency resolution.

```python
# The global connection factory that creates and cleans up vanilla
# connections.
registry.register_factory(Connection, engine.connect)

# Create a type alias with an idiomatic name.
type ConnectionWithUserID = Connection

def middleware(request):
    def set_user_id(svcs_container):
        conn = svcs_container.get(Connection)
        conn.execute(text("SET user = :id", {"id": user_id}))

        return conn

    # Use a factory to keep the service lazy. If the view never asks for a
    # connection, we never connect -- or set a user.
    container.register_local_factory(ConnectionWithUserID, set_user_id)
```

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

### Service Layer Function Example

Source: https://github.com/hynek/svcs/blob/main/docs/glossary.md

Illustrates a service layer function that coordinates database operations, external queues, and domain model logic. Services like 'unit_of_work' and 'mail_q' are injected as parameters.

```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()
```

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

### Health Check Endpoint Example (Flask)

Source: https://github.com/hynek/svcs/blob/main/docs/core-concepts.md

Example of a health check endpoint implementation using Flask. This snippet demonstrates how to retrieve the container from the request context and perform health checks.

```python
from flask import Flask, g, jsonify
import svcs

def health_check():
    container: svcs.Container = g.svcs_container
    pings = container.get_pings()
    return jsonify({
        "services": [
            {
                "name": ping.name,
                "ok": True,
                "error": None
            }
            for ping in pings
        ]
    })

def setup_health_check(app: Flask):
    app.add_url_rule("/health", "health_check", health_check)
```

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

### Initialize Svcs App with Flask

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/flask.md

Initialize the Svcs application registry within a Flask application factory. This setup allows for automatic cleanup of resources like database connections when the application shuts down.

```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),
        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

```

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

### Get Service from Current Request

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/pyramid.md

Use svcs.pyramid.get(request, ServiceType) for a more direct way to acquire a service from the current request's container. This is a convenient shortcut for common use cases.

```python
import svcs

def view(request):
    db = svcs.pyramid.get(request, Database)
```

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

### Get Database Connection on Demand

Source: https://github.com/hynek/svcs/blob/main/docs/glossary.md

Acquire a database connection only when the form is valid. This avoids unnecessary resource allocation and demonstrates service location within a request context.

```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()
```

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

### Health Check Endpoint with AIOHTTP

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/aiohttp.md

Example of a health check endpoint using svcs.aiohttp.get_pings. Ensure services are registered with the registry for this to function correctly.

```python
from aiohttp import web
import svcs

async def health_check(request: web.Request) -> web.Response:
    pings = await svcs.aiohttp.get_pings(request)
    return web.json_response(pings)


app = web.Application()
app.router.add_get("/health", health_check)

```

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

### Health Check Endpoint Example

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/pyramid.md

Implement a health check endpoint in your Pyramid application using svcs.pyramid.get_pings. This allows you to monitor the health of your services.

```python
from pyramid.view import view_config

@view_config(route_name='health')
def health_check(request):
    return svcs.pyramid.get_pings(request)
```

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

### Acquire svcs Container from Request

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/pyramid.md

Access the request-scoped svcs.Container using svcs_from(request) to retrieve services. This is the recommended way to get a container within a Pyramid request context.

```python
from svcs.pyramid import svcs_from

def view(request):
    db = svcs_from(request).get(Database)
```

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

### Acquire Service using Container Proxy

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/flask.md

Access services registered with Svcs within a Flask application using the `container` local proxy. This provides a more concise way to get services compared to `svcs.flask.get`.

```python
from flask import Flask
from svcs.flask import container
from sqlalchemy import Connection

app = Flask(__name__)

@app.get("/")
def index() -> flask.ResponseValue:
    conn = container.get(Connection)

```

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

### Initialization

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/pyramid.md

Initialize svcs with a Pyramid configurator and register services.

```APIDOC
## svcs.pyramid.init(configurator, *, tween_under=None, tween_over=None)

### Description
Initializes the svcs integration for a Pyramid application.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **configurator** (pyramid.config.Configurator) - The Pyramid configurator object.
- **tween_under** (str, optional) - The name of the tween to place the svcs tween under.
- **tween_over** (str, optional) - The name of the tween to place the svcs tween over.

### Request Example
```python
from pyramid.config import Configurator
import svcs.pyramid

def make_app():
    settings = {}
    with Configurator(settings=settings) as config:
        svcs.pyramid.init(config)
        # ... other configurations ...
        return config.make_wsgi_app()
```

### Response
None
```

```APIDOC
## svcs.pyramid.register_factory(configurator, svc_type, factory, *, name=None)

### Description
Registers a factory for a service type with the Pyramid configurator.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **configurator** (pyramid.config.Configurator) - The Pyramid configurator object.
- **svc_type** (type) - The type of the service to register.
- **factory** (callable) - The factory function that creates the service.
- **name** (str, optional) - An optional name for the service.

### Request Example
```python
from pyramid.config import Configurator
import svcs.pyramid

def db_factory():
    # ... create database connection ...
    return Database()

with Configurator() as config:
    svcs.pyramid.init(config)
    svcs.pyramid.register_factory(config, Database, db_factory)
    # ...
```

### Response
None
```

```APIDOC
## svcs.pyramid.register_value(configurator, svc_type, value, *, name=None)

### Description
Registers a pre-created value for a service type with the Pyramid configurator.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **configurator** (pyramid.config.Configurator) - The Pyramid configurator object.
- **svc_type** (type) - The type of the service to register.
- **value** (any) - The pre-created service value.
- **name** (str, optional) - An optional name for the service.

### Request Example
```python
from pyramid.config import Configurator
import svcs.pyramid

my_db_connection = create_db_connection()

with Configurator() as config:
    svcs.pyramid.init(config)
    svcs.pyramid.register_value(config, Database, my_db_connection)
    # ...
```

### Response
None
```

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

### Basic Container Usage

Source: https://github.com/hynek/svcs/blob/main/docs/core-concepts.md

Demonstrates creating a container, acquiring services, and observing instance caching. Services are cached by default, ensuring the same instance is returned on subsequent calls.

```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'
```

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

### Registering Factories and Values in a Registry

Source: https://github.com/hynek/svcs/blob/main/docs/core-concepts.md

Demonstrates how to create a registry and register factories for types (e.g., uuid.uuid4 for UUID) and direct values (e.g., 'Hello World' for str). Shows how to check if a type is registered.

```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
```

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

### Starlette Application Initialization

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/starlette.md

Demonstrates how to initialize a Starlette application with svcs, including setting up the lifespan and middleware.

```APIDOC
## Initialization

To use *svcs* with Starlette, you have to pass a [*lifespan*](https://www.starlette.io/lifespan/) -- that has been wrapped by {class}`svcs.starlette.lifespan` -- and a {class}`~svcs.starlette.SVCSMiddleware` to your application:

```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=[...],
)
```
```

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

### Initialization

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/aiohttp.md

Initialize the svcs integration with your aiohttp application.

```APIDOC
## init_app

### Description
Initializes the svcs integration for an aiohttp application.

### Function Signature
`svcs.aiohttp.init_app(app: aiohttp.web.Application)`

### Parameters
- **app** (aiohttp.web.Application) - The aiohttp application instance.
```

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

### Dependency Injection via Parameters

Source: https://github.com/hynek/svcs/blob/main/docs/glossary.md

Demonstrates dependency injection by passing required services (smtp, db) as parameters to the function, improving testability and decoupling.

```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)
```

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

### Testing with Dependency Replacement

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/starlette.md

Illustrates how to replace dependencies in tests to isolate components and test error handling.

```APIDOC
## Testing

The centralized service registry makes it straight-forward to selectively replace dependencies within your application in tests even if you have many dependencies to handle.

Let's take this simple application as an example:

```{literalinclude} ../examples/starlette/simple_starlette_app.py
```

Now if you want to make a request against the `get_user` view, but want the database to raise an error to see if it's properly handled, you can do this:

```{literalinclude} ../examples/starlette/test_simple_starlette_app.py
```

As you can see, we can inspect the decorated lifespan function to get the registry that got injected and you can overwrite it later.

::: {important}
    You must overwrite *after* the application has been initialized.
    Otherwise the lifespan function overwrites your settings.
:::

```

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

### Service Acquisition using svcs_from

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/starlette.md

Shows how to acquire services from the request context using the `svcs_from` helper function.

```APIDOC
## Service Acquisition using svcs_from

You can either use {func}`svcs.starlette.svcs_from`:

```python
from svcs.starlette import svcs_from

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

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

### Registering a Factory with SQLAlchemy Connection

Source: https://github.com/hynek/svcs/blob/main/docs/why.md

Demonstrates registering a SQLAlchemy connection factory with a health check and a cleanup callback. The factory uses a generator to yield connections.

```python
import atexit

from sqlalchemy import Connection, create_engine, text

...

engine = create_engine("postgresql://localhost")

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

registry = svcs.Registry()
registry.register_factory(
    Connection,
    connection_factory,
    ping=lambda conn: conn.execute(text("SELECT 1")), # ← health check
    on_registry_close=engine.dispose
)

@atexit.register
def cleanup():
    registry.close()  # calls engine.dispose()
```

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

### Registering Cleanup Callback for a Factory

Source: https://github.com/hynek/svcs/blob/main/docs/core-concepts.md

Illustrates how to register a callback function to be executed when the registry is closed, typically used for resource cleanup like disposing of database engines.

```python
engine = create_engine(url)

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

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

### Initialize Starlette App with svcs

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/starlette.md

Configure your Starlette application with the svcs lifespan and middleware for dependency injection and lifecycle management.

```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=[...],
)
```

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

### Container as Context Manager with Cleanup

Source: https://github.com/hynek/svcs/blob/main/docs/core-concepts.md

Illustrates using a container as an async context manager for automatic cleanup. Factories yielding instances are automatically wrapped in context managers, ensuring cleanup actions are executed upon exiting the 'with' block.

```python
>>> reg = svcs.Registry()
>>> def clean_factory() -> str:
...     yield "Hello World"
...     print("Cleaned up!")
>>> reg.register_factory(str, clean_factory)
>>> with svcs.Container(reg) as con:
...     _ = con.get(str)
Cleaned up!
```

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

### Registering Multiple Factories for the Same Type

Source: https://github.com/hynek/svcs/blob/main/docs/core-concepts.md

Shows how to register multiple factories for the same base type by creating distinct subclasses or using `typing.Annotated`. This is useful for scenarios like multiple database connections.

```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)

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

# This works with Mypy, but NOT with Pyright:
SecondaryConnection = Annotated[Connection, "secondary"]

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

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

### Registering SQLAlchemy Connection as Context Manager

Source: https://github.com/hynek/svcs/blob/main/docs/why.md

Shows an equivalent registration of a SQLAlchemy connection using its context manager directly. This avoids the need for a separate generator factory.

```python
engine = create_engine("postgresql://localhost")

registry = svcs.Registry()
registry.register_factory(
    Connection,
    engine.connect,  # ← sqlalchemy.Connection is a context manager
    ping=lambda conn: conn.execute(text("SELECT 1")),
    on_registry_close=engine.dispose
)

@atexit.register
def cleanup():
    registry.close()
```

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

### Classic Service Location in Service Layer

Source: https://github.com/hynek/svcs/blob/main/docs/glossary.md

Illustrates traditional service location within the service layer, which is generally not recommended due to testability and reasoning challenges.

```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)
```

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

### Service Acquisition using aget

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/starlette.md

Demonstrates acquiring services directly using the `svcs.starlette.aget` function.

```APIDOC
## Service Acquisition using aget

Or you can use {func}`svcs.starlette.aget` to extract your services directly:

```python
import svcs

async def view(request):
    db = await svcs.starlette.aget(request, Database)
```
```

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

### Testing FastAPI Service Replacement

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/fastapi.md

Demonstrates how to replace a service dependency in a FastAPI application during testing by inspecting and overwriting the registry provided by the lifespan function.

```python
from fastapi.testclient import TestClient


class MockDatabase:
    async def connect(self):
        raise ConnectionError("Cannot connect to DB")


client = TestClient(app)


# Inspect the lifespan function to get the registry
registry = app.state.svcs_registry

# Overwrite the Database service with a mock
registry.overwrite_factory(Database, MockDatabase)


response = client.get("/users/1")
assert response.status_code == 500
assert response.json() == {"detail": "Cannot connect to DB"}
```

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

### Initialize svcs in Pyramid

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/pyramid.md

Use svcs.pyramid.init() to set up the svcs registry and tween within your Pyramid application's configurator. You can also register factories and values using helper functions.

```python
from pyramid.config import Configurator
import svcs

def make_app():
    ...

    with Configurator(settings=settings) as config:
        svcs.pyramid.init(config)
        svcs.pyramid.register_factory(config, Database, db_factory)

        ...

        return config.make_wsgi_app()
```

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

### Debugging Registrations

Source: https://github.com/hynek/svcs/blob/main/docs/core-concepts.md

Learn how to enable debug logging in svcs to trace service registrations and identify factory definitions.

```APIDOC
## Debugging Registrations

If you are confused about where a particular factory for a type has been defined, *svcs* logs every registration at debug level along with a stack trace.

Set the *svcs* logger to `DEBUG` to see them:

```python
# Example of enabling debug logging
# (Actual code would be in a literalinclude block)
import logging

logging.basicConfig()
logging.getLogger('svcs').setLevel(logging.DEBUG)
```

This will provide output detailing each service registration and its corresponding stack trace, helping you pinpoint the origin of factories and values.
```

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

### FastAPI Service Locator Integration

Source: https://github.com/hynek/svcs/blob/main/docs/index.md

Integrate svcs with FastAPI using DepContainer for dependency injection. Services are retrieved asynchronously.

```python
import svcs

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

    ...
```

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

### Container with Factory and Recursive Acquisition

Source: https://github.com/hynek/svcs/blob/main/docs/core-concepts.md

Shows how to register a factory function that can recursively acquire services from the container. The container instance is passed as the first argument to the factory if it's named 'svcs_container' or annotated.

```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'
```

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

### Manual Dependency Construction

Source: https://github.com/hynek/svcs/blob/main/docs/glossary.md

Shows a function that manually constructs its dependencies (SmtpSender, DbConnection), making it difficult to test as it performs external actions like sending emails.

```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)
```

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

### AIOHTTP Service Locator Integration

Source: https://github.com/hynek/svcs/blob/main/docs/index.md

Use svcs.aiohttp.aget to retrieve services within an aiohttp request context. Ensure services are awaited.

```python
import svcs

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

    ...
```

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

### Acquire Services within a Request Context

Source: https://github.com/hynek/svcs/blob/main/README.md

Acquire multiple services (Database, WebAPIClient, Cache) from the request context. Services are automatically cleaned up when the request ends.

```python
from svcs.your_framework import svcs_from

def view(request):
    db, api, cache = svcs_from(request).get(Database, WebAPIClient, Cache)
```

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

### Register Services using Re-exported Helpers

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/flask.md

Register services within an application factory using a custom module that re-exports Svcs Flask helpers. This pattern centralizes service registration logic.

```python
from your_app import services
from sqlalchemy import Connection

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

    return app

```

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

### Acquire Service using aget

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/starlette.md

Directly acquire a service using svcs.starlette.aget with the request object.

```python
import svcs

async def view(request):
    db = await svcs.starlette.aget(request, Database)
```

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

### Registering Services

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/aiohttp.md

Register service factories or values with the application.

```APIDOC
## register_factory

### Description
Registers a service factory with the application's registry.

### Function Signature
`svcs.aiohttp.register_factory(app: aiohttp.web.Application, svc_type: type, factory: callable)`

### Parameters
- **app** (aiohttp.web.Application) - The aiohttp application instance.
- **svc_type** (type) - The type of the service to register.
- **factory** (callable) - The factory function that creates the service.
```

```APIDOC
## register_value

### Description
Registers a service value with the application's registry.

### Function Signature
`svcs.aiohttp.register_value(app: aiohttp.web.Application, svc_type: type, value: object)`

### Parameters
- **app** (aiohttp.web.Application) - The aiohttp application instance.
- **svc_type** (type) - The type of the service to register.
- **value** (object) - The service instance to register.
```

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

### Service Locator in Composition Root

Source: https://github.com/hynek/svcs/blob/main/docs/glossary.md

Uses 'svcs' as a service locator within the composition root to acquire a Database service, which is then injected into the service layer function.

```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()
```

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

### Debug Service Registrations with Logging

Source: https://github.com/hynek/svcs/blob/main/docs/core-concepts.md

Set the svcs logger to DEBUG to view detailed registration information, including stack traces, which helps in understanding where factories and values are defined.

```python
from svcs import Container
import logging
import datetime

logging.basicConfig()
log = logging.getLogger(__name__)
log.setLevel(logging.DEBUG)

reg = Container()
reg.register_factory(datetime, datetime.now)
reg.register_value(str, "Hello World")

```

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

### Testing Fixtures

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/pyramid.md

Pytest fixtures for testing Pyramid applications with svcs.

```APIDOC
## app fixture

### Description
Provides a WebTest application instance configured with svcs for testing.

### Parameters
None

### Request Example
```python
import pytest
import webtest

# Assuming your_app.make_app() initializes svcs

@pytest.fixture
def app():
    app = your_app.make_app()
    with svcs.pyramid.get_registry(app):
        yield webtest.TestApp(app)
```

### Response
- **webtest.TestApp** - A configured WebTest application.
```

```APIDOC
## registry fixture

### Description
Provides the svcs registry from a Pyramid application for testing overrides.

### Parameters
None

### Request Example
```python
import pytest

@pytest.fixture
def registry(app):
    return svcs.pyramid.get_registry(app)

# Example usage in a test:
def test_override_database(app, registry):
    from sqlalchemy import Engine
    from unittest.mock import Mock

    mock_engine = Mock(spec_set=Engine)
    mock_engine.execute.side_effect = RuntimeError("Boom!")

    registry.register_value(Engine, mock_engine)
    # Now tests using 'app' will use the mocked Engine
```

### Response
- **svcs.Registry** - The svcs registry of the application.
```

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

### Pyramid Service Locator Integration

Source: https://github.com/hynek/svcs/blob/main/docs/index.md

Integrate svcs with Pyramid using svcs.pyramid.get to retrieve services within a Pyramid request. Requires the view_config decorator.

```python
import svcs

@view_config(route_name="index")
def view(request):
    db, api, cache = svcs.pyramid.get(request, Database, WebAPIClient, Cache)

    ...
```

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

### Registry Class API

Source: https://github.com/hynek/svcs/blob/main/docs/core-concepts.md

Reference for the `Registry` class, including methods for registering factories and values, and managing the registry lifecycle.

```APIDOC
## API Reference - Registry

### `Registry()`

Represents a registry for managing service factories and values.

#### Methods

- **`register_factory(type, factory, **kwargs)`**: Registers a factory function for a given service type.
- **`register_value(type, value, **kwargs)`**: Registers a specific value for a given service type.
- **`close()`**: Closes the registry and releases resources.
- **`aclose()`**: Asynchronously closes the registry.
- **`__contains__(type)`**: Checks if a service type is registered.
- **`__iter__()`**: Returns an iterator over the registered service types.
```

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

### Flask Service Locator Integration

Source: https://github.com/hynek/svcs/blob/main/docs/index.md

Use svcs.flask.get to retrieve services within a Flask request context. This is a synchronous retrieval.

```python
import svcs

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

    ...
```

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

### Service Acquisition

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/flask.md

Utilities for acquiring services within a Flask application context.

```APIDOC
## Service Acquisition

### `svcs_from(app)`

Returns the service container for the given Flask application.

### `container`

A `werkzeug.local.LocalProxy` that transparently calls `svcs_from()` for you when accessed within a request context.

### `get(svc_types)`

Same as `svcs.Container.get()`, but uses the container from `flask.g`.

### `get_abstract(svc_types)`

Retrieves abstract service types.

### `get_pings()`

Retrieves all registered service ping functions.
```

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

### Cleanup

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/aiohttp.md

Clean up the registry when the aiohttp application is closing.

```APIDOC
## aclose_registry

### Description
Closes the svcs registry and runs any registered on_registry_close callbacks.
This is typically called automatically during application shutdown.

### Function Signature
`svcs.aiohttp.aclose_registry(app: aiohttp.web.Application)`

### Parameters
- **app** (aiohttp.web.Application) - The aiohttp application instance.
```

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

### Global Factory Depending on Local Services

Source: https://github.com/hynek/svcs/blob/main/docs/core-concepts.md

While possible, it's generally safer to avoid global factories depending on local ones to prevent issues with service caching and initialization order.

```python
from sqlalchemy import text

def connect_and_set_user(svcs_container):
    user_id = svcs_container.get(User).user_id
    with engine.connect() as conn:
        conn.execute(text("SET user = :id", {"id": user_id}))

        yield conn

registry.register_factory(Connection, connect_and_set_user)
```

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

### Local Factory for Request-Specific Services

Source: https://github.com/hynek/svcs/blob/main/docs/core-concepts.md

Define a local factory for request-specific services, ensuring dependencies are met within the request context. This pattern helps manage service lifecycles and dependencies more effectively.

```python
def middleware(request):
    container.register_local_value(User, User(request.user_id, ...))
```

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

### Registering Services

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/flask.md

Functions for registering service factories and values with the service registry.

```APIDOC
## Registering Services

### `register_factory(svc_type, factory)`

Registers a factory function for a given service type.

### `register_value(svc_type, value)`

Registers a pre-instantiated value for a given service type.
```

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

### Access Registry and Container via Thread Locals

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/pyramid.md

While discouraged, you can access the Pyramid registry and the current request's svcs container using thread-local functions. These functions are convenient but only work within active Pyramid requests.

```python
import svcs

def view(request):
    registry = svcs.pyramid.get_registry()
    container = svcs.pyramid.svcs_from()
```

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

### FastAPI Application Initialization with svcs Lifespan

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/fastapi.md

Initialize your FastAPI application with a lifespan that registers services with the svcs registry. The registry is stored in the request state and automatically closed when the app is done.

```python
from fastapi import FastAPI

import svcs


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

    yield {"your": "other", "initial": "state"}

    # Registry is closed automatically when the app is done.


app = FastAPI(lifespan=lifespan)
```

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

### Testing Utilities

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/flask.md

Functions for testing service registrations within a Flask application. These should not be used in production.

```APIDOC
## Testing

:::caution
These functions should not be used in production code.

They always reset the container and run all cleanups when overwriting a service.

See also {ref}`flask-testing`.
:::

### `overwrite_factory(svc_type, factory)`

Overwrites a registered service factory with a new one for testing purposes.

### `overwrite_value(svc_type, value)`

Overwrites a registered service value with a new one for testing purposes.
```

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

### Container Class API

Source: https://github.com/hynek/svcs/blob/main/docs/core-concepts.md

Reference for the `Container` class, which provides methods for retrieving services, managing local registrations, and checking service health.

```APIDOC
## API Reference - Container

### `Container()`

Represents a container for resolving services.

#### Methods

- **`get(type)`**: Synchronously retrieves an instance of the service type.
- **`aget(type)`**: Asynchronously retrieves an instance of the service type.
- **`get_abstract(type)`**: Retrieves the factory or value for an abstract service type.
- **`aget_abstract(type)`**: Asynchronously retrieves the factory or value for an abstract service type.
- **`register_local_factory(type, factory, **kwargs)`**: Registers a local factory within the container.
- **`register_local_value(type, value, **kwargs)`**: Registers a local value within the container.
- **`close()`**: Closes the container and releases resources.
- **`aclose()`**: Asynchronously closes the container.
- **`get_pings()`**: Retrieves all registered service ping objects.
- **`__contains__(type)`**: Checks if a service type is present in the container.
```

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

### Testing Starlette App with Dependency Overrides

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/starlette.md

Replace dependencies in tests by accessing and modifying the registry via the decorated lifespan function. Ensure overrides are applied after initialization.

```python
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.routing import Route
from starlette.testclient import TestClient

import svcs


class Database:
    async def connect(self):
        pass

    async def disconnect(self):
        pass

    async def query(self, query: str):
        return query


async def get_user(request):
    db = await svcs.starlette.aget(request, Database)
    result = await db.query("SELECT * FROM users")
    return JSONResponse({"user": result})


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


app = Starlette(
    lifespan=lifespan,
    middleware=[Middleware(svcs.starlette.SVCSMiddleware)],
    routes=[
        Route("/user", endpoint=get_user),
    ],
)


client = TestClient(app)


def test_get_user_db_error():
    # Overwrite after initialization
    app.state.svcs_registry.register_value(Database, None)  # type: ignore

    response = client.get("/user")
    assert response.status_code == 500

```

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

### Override Service in Test

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/pyramid.md

Demonstrates how to override a registered service (e.g., Engine) within a Pyramid test using the registry fixture. This is useful for isolating components and testing error conditions.

```python
from sqlalchemy import Engine
from unittest.mock import Mock

def test_broken_database(app, registry):
    boom = Mock(spec_set=Engine)
    boom.execute.side_effect = RuntimeError("Boom!")

    registry.register_value(Engine, boom)  # ← override the database

    resp = app.get("/some-url")

    assert 500 == resp.status_code
```

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

### API Reference: aget_abstract

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/starlette.md

Reference for the `aget_abstract` function, used for acquiring abstract services.

```APIDOC
```{eval-rst}
.. autofunction:: aget_abstract
```
```

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

### Acquire Connection in Flask View

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/flask.md

Retrieve a registered service, such as a database Connection, within a Flask view function using `svcs.flask.get`. This assumes the service has been previously registered.

```python
from flask import Flask
import svcs
from sqlalchemy import Connection

app = Flask(__name__)

@app.get("/")
def index() -> flask.ResponseValue:
    conn = svcs.flask.get(Connection)

```

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

### Create Module for Flask Helpers

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/flask.md

Simplify service access in Flask views by creating a module that re-exports common Svcs Flask helpers. This centralizes imports and makes view code cleaner.

```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",
]

```

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

### Service Acquisition

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/pyramid.md

Acquire services from the request-scoped container in Pyramid.

```APIDOC
## svcs.pyramid.svcs_from(request)

### Description
Retrieves the request-scoped svcs container from a Pyramid request object.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **request** (pyramid.request.Request) - The Pyramid request object.

### Request Example
```python
from svcs.pyramid import svcs_from

def view(request):
    db = svcs_from(request).get(Database)
    # ...
```

### Response
- **svcs.Container** - The request-scoped svcs container.
```

```APIDOC
## svcs.pyramid.get(request, *svc_types)

### Description
Retrieves services from the current request's svcs container.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **request** (pyramid.request.Request) - The Pyramid request object.
- **svc_types** (type) - One or more service types to retrieve.

### Request Example
```python
import svcs

def view(request):
    db = svcs.pyramid.get(request, Database)
    # ...
```

### Response
- **svcs.Container** - The requested service(s).
```

```APIDOC
## svcs.pyramid.get_registry(obj=None)

### Description
Retrieves the svcs registry associated with a Pyramid object (configurator, app, or request).

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **obj** (object, optional) - The Pyramid object (Configurator, WSGI app, or Request) to get the registry from. If None, attempts to use thread-local context.

### Request Example
```python
# Assuming 'request' is an active Pyramid request object
registry = svcs.pyramid.get_registry(request)
```

### Response
- **svcs.Registry** - The svcs registry.
```

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

### Pyramid Testing Fixtures

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/pyramid.md

Use pytest fixtures to obtain a WebTest application and its registry for testing. This allows you to override services and simulate application behavior.

```python
from your_app import make_app

import pytest
import svcs
import webtest

@pytest.fixture
def app():
    app = make_app()

    with svcs.pyramid.get_registry(app):
        yield webtest.TestApp(app)


@pytest.fixture
def registry(app):
    return svcs.pyramid.get_registry(app)
```

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

### API Reference: aget

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/starlette.md

Reference for the `aget` function, which retrieves services asynchronously from the request context.

```APIDOC
## API Reference

### Service Acquisition

```{eval-rst}
.. function:: aget(request: starlette.requests.Request, svc_type1: type, ...)
   :async:

   Same as :meth:`svcs.Container.aget`, but uses the container from *request*.
```
```

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

### API Reference: svcs_from

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/starlette.md

Reference for the `svcs_from` function, which provides access to the service container from the request.

```APIDOC
```{eval-rst}
.. autofunction:: svcs_from
```
```

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

### Service Acquisition

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/aiohttp.md

Acquire services from the request or application context.

```APIDOC
## svcs_from

### Description
Retrieves the svcs.Container from the aiohttp request.

### Function Signature
`svcs.aiohttp.svcs_from(request: aiohttp.web.Request) -> svcs.Container`

### Parameters
- **request** (aiohttp.web.Request) - The incoming aiohttp request.
```

```APIDOC
## aget

### Description
Acquires a service instance from the request's container.
This is an asynchronous function.

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

### Parameters
- **request** (aiohttp.web.Request) - The incoming aiohttp request.
- **svc_type1** (type) - The type of the service to acquire.
```

```APIDOC
## aget_abstract

### Description
Acquires an abstract service instance from the request's container.
This is an asynchronous function.

### Function Signature
`async def aget_abstract(request: aiohttp.web.Request, svc_type1: type, ...)`

### Parameters
- **request** (aiohttp.web.Request) - The incoming aiohttp request.
- **svc_type1** (type) - The abstract type of the service to acquire.
```

```APIDOC
## get_pings

### Description
Retrieves all registered ping functions from the request's container.

### Function Signature
`svcs.aiohttp.get_pings(request: aiohttp.web.Request) -> dict`

### Parameters
- **request** (aiohttp.web.Request) - The incoming aiohttp request.
```

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

### Health Checks

Source: https://github.com/hynek/svcs/blob/main/docs/integrations/starlette.md

Explains how to implement health checks in a Starlette application using `svcs_from` or `get_pings`.

```APIDOC
## Health Checks

As with services, you have the option to either {func}`svcs.starlette.svcs_from` on the request or go straight for {func}`svcs.starlette.get_pings`.

A health endpoint could look like this:

```{literalinclude} ../examples/starlette/health_check.py
```
```

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

### Starlette Service Locator Integration

Source: https://github.com/hynek/svcs/blob/main/docs/index.md

Use svcs.starlette.aget to retrieve services within a Starlette request context. This is an asynchronous retrieval.

```python
import svcs

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

    ...
```