### Getting Started with FastAPI Injection

Source: https://github.com/woltapp/magic-di/blob/main/README.md

Demonstrates how to use inject_app to manage dependency lifecycles and Provide for dependency injection in FastAPI endpoints.

```python
from fastapi import FastAPI
from magic_di import Connectable
from magic_di.fastapi import inject_app, Provide

app = inject_app(FastAPI())

class Database:
    connected: bool = False
    def __connect__(self):
        self.connected = True
    def __disconnect__(self):
        self.connected = False

class Service(Connectable):
    def __init__(self, db: Database):
        self.db = db
    def is_connected(self):
        return self.db.connected

@app.get(path="/hello-world")
def hello_world(service: Provide[Service]) -> dict:
    return {"is_connected": service.is_connected()}
```

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

### Development Environment Setup

Source: https://github.com/woltapp/magic-di/blob/main/README.md

Standard commands for setting up the development environment using Poetry and running tests.

```shell
poetry install --all-extras
poetry shell
pytest
```

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

### Install Magic-DI via pip

Source: https://github.com/woltapp/magic-di/blob/main/README.md

Commands to install the core magic-di library and optional integrations for FastAPI and Celery.

```bash
pip install magic-di
pip install 'magic-di[fastapi]'
pip install 'magic-di[celery]'
```

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

### Mock Dependencies for Testing

Source: https://github.com/woltapp/magic-di/blob/main/README.md

Provides examples of mocking dependencies in tests using the injector.override context manager. This includes using the built-in InjectableMock or custom mock classes.

```python
import pytest
from magic_di.testing import InjectableMock

@pytest.fixture()
def client(injector: DependencyInjector, service_mock: InjectableMock):
    with injector.override({Service: service_mock.mock_cls}):
        with TestClient(app) as client:
            yield client
```

```python
from magic_di.testing import get_injectable_mock_cls

@pytest.fixture()
def client(injector: DependencyInjector, service_mock: Service):
    with injector.override({Service: get_injectable_mock_cls(service_mock)}):
        with TestClient(app) as client:
            yield client
```

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

### Python: Binding Interfaces to Implementations with DependencyInjector

Source: https://context7.com/woltapp/magic-di/llms.txt

Shows how to use the `DependencyInjector.bind()` method to associate an interface (defined using `typing.Protocol`) with a concrete implementation. This allows for programming to interfaces, enabling easier swapping of implementations. The example binds `CacheInterface` to `RedisCache`.

```python
from typing import Protocol
from magic_di import DependencyInjector, Connectable

class CacheInterface(Protocol):
    async def get(self, key: str) -> str | None: ...
    async def set(self, key: str, value: str) -> None: ...

class RedisCache(Connectable):
    async def __connect__(self) -> None:
        print("Redis connected")

    async def __disconnect__(self) -> None:
        print("Redis disconnected")

    async def get(self, key: str) -> str | None:
        return f"redis_value_{key}"

    async def set(self, key: str, value: str) -> None:
        print(f"Set {key}={value} in Redis")

class UserService(Connectable):
    def __init__(self, cache: CacheInterface):
        self.cache = cache

    async def get_cached_user(self, user_id: str) -> str | None:
        return await self.cache.get(f"user:{user_id}")

# Bind interface to implementation
injector = DependencyInjector()
injector.bind({CacheInterface: RedisCache})

service = injector.inject(UserService)()
# service.cache will be an instance of RedisCache
```

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

### Celery Integration with Magic-DI Loaders and Tasks

Source: https://context7.com/woltapp/magic-di/llms.txt

Shows how to configure Celery with Magic-DI's custom loader and task class for dependency injection and async support. It includes examples for both function-based and class-based Celery tasks.

```python
from celery import Celery
from magic_di import Connectable
from magic_di.celery import get_celery_loader, InjectableCeleryTask, PROVIDE

app = Celery(
    loader=get_celery_loader(),
    task_cls=InjectableCeleryTask,
)

class Calculator(Connectable):
    async def __connect__(self) -> None:
        print("Calculator initialized")

    async def __disconnect__(self) -> None:
        print("Calculator shutdown")

    async def calculate(self, x: int, y: int) -> int:
        return x + y

# Function-based task with dependency injection
@app.task
async def add_numbers(x: int, y: int, calculator: Calculator = PROVIDE) -> int:
    result = await calculator.calculate(x, y)
    return result

# Usage: add_numbers.delay(5, 3)
```

```python
from dataclasses import dataclass
from celery import Celery
from magic_di import Connectable
from magic_di.celery import (
    get_celery_loader,
    InjectableCeleryTask,
    BaseCeleryConnectableDeps,
    PROVIDE,
)

app = Celery(
    loader=get_celery_loader(),
    task_cls=InjectableCeleryTask,
)

class Database(Connectable):
    async def __connect__(self) -> None:
        print("DB connected")

    async def __disconnect__(self) -> None:
        print("DB disconnected")

    async def save(self, data: dict) -> int:
        return 1  # Return saved record ID

class NotificationService(Connectable):
    async def __connect__(self) -> None:
        pass

    async def __disconnect__(self) -> None:
        pass

    async def notify(self, message: str) -> None:
        print(f"Notification: {message}")

@dataclass
class OrderTaskDeps(BaseCeleryConnectableDeps):
    db: Database
    notifier: NotificationService

class ProcessOrderTask(InjectableCeleryTask):
    deps: OrderTaskDeps

    async def run(self, order_data: dict) -> int:
        order_id = await self.deps.db.save(order_data)
        await self.deps.notifier.notify(f"Order {order_id} processed")
        return order_id

app.register_task(ProcessOrderTask())
# Usage: ProcessOrderTask().delay({"item": "widget", "qty": 5})
```

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

### Initialize Redis with Zero Configuration

Source: https://github.com/woltapp/magic-di/blob/main/README.md

Demonstrates how to use Pydantic settings to automatically fetch Redis configuration from environment variables without manual injection.

```python
from dataclasses import dataclass, field
from pydantic import Field
from pydantic_settings import BaseSettings
from redis.asyncio import Redis as RedisClient, from_url

class RedisConfig(BaseSettings):
    url: str = Field(validation_alias='REDIS_URL')
    decode_responses: bool = Field(validation_alias='REDIS_DECODE_RESPONSES')

@dataclass
class Redis:
    config: RedisConfig = field(default_factory=RedisConfig)
    client: RedisClient = field(init=False)

    async def __connect__(self):
        self.client = await from_url(self.config.url, decode_responses=self.config.decode_responses)
        await self.client.ping()

    async def __disconnect__(self):
        await self.client.close()

    @property
    def db(self) -> RedisClient:
        return self.client

Redis()
```

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

### Python: Implement Connectable for Redis Client

Source: https://context7.com/woltapp/magic-di/llms.txt

Demonstrates how to create a Redis client dependency using the `Connectable` base class. It includes configuration loading from environment variables using Pydantic, automatic connection establishment via `__connect__`, and proper disconnection via `__disconnect__`. The `db` property provides access to the underlying Redis client.

```python
from dataclasses import dataclass, field
from pydantic import Field
from pydantic_settings import BaseSettings
from redis.asyncio import Redis as RedisClient, from_url
from magic_di import Connectable

class RedisConfig(BaseSettings):
    url: str = Field(validation_alias='REDIS_URL')
    decode_responses: bool = Field(validation_alias='REDIS_DECODE_RESPONSES')

@dataclass
class Redis(Connectable):
    config: RedisConfig = field(default_factory=RedisConfig)
    client: RedisClient = field(init=False)

    async def __connect__(self) -> None:
        self.client = await from_url(
            self.config.url,
            decode_responses=self.config.decode_responses
        )
        await self.client.ping()

    async def __disconnect__(self) -> None:
        await self.client.close()

    @property
    def db(self) -> RedisClient:
        return self.client
```

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

### Perform Custom Dependency Injection

Source: https://github.com/woltapp/magic-di/blob/main/README.md

Demonstrates how to integrate magic-di using the helper function inject_and_run or by manually managing the DependencyInjector lifecycle. This is useful for bootstrapping applications with injected dependencies.

```python
from magic_di.utils import inject_and_run

async def main(worker: Worker):
    await worker.run()

if __name__ == '__main__':
    inject_and_run(main)
```

```python
import asyncio
from magic_di import DependencyInjector

async def run_worker(worker: Worker):
    await worker.run()

async def main():
    injector = DependencyInjector()
    injected_fn = injector.inject(run_worker)
    async with injector:
        await injected_fn()

if __name__ == '__main__':
    asyncio.run(main())
```

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

### Override Bindings with Context Manager in Python

Source: https://context7.com/woltapp/magic-di/llms.txt

Demonstrates how to temporarily replace dependency bindings using the `override()` context manager for testing. It shows how to mock a service and ensure original bindings are restored upon exiting the context.

```python
from magic_di import DependencyInjector, Connectable
from magic_di.testing import InjectableMock

class EmailService(Connectable):
    async def send_email(self, to: str, subject: str) -> bool:
        # Real email sending logic
        return True

class NotificationService(Connectable):
    def __init__(self, email: EmailService):
        self.email = email

    async def notify_user(self, user_email: str) -> bool:
        return await self.email.send_email(user_email, "Notification")

# In tests, override with mock
injector = DependencyInjector()

mock_email = InjectableMock()
mock_email.send_email.return_value = True

with injector.override({EmailService: mock_email.mock_cls}):
    service = injector.inject(NotificationService)()
    # service.email is now the mock
```

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

### Inject Dependencies with DependencyInjector

Source: https://github.com/woltapp/magic-di/blob/main/README.md

Shows how to use the DependencyInjector to manage object lifecycles and inject configurations explicitly.

```python
from dataclasses import dataclass, field
from pydantic import Field
from pydantic_settings import BaseSettings
from redis.asyncio import Redis as RedisClient, from_url
from magic_di import Connectable, DependencyInjector

class RedisConfig(Connectable, BaseSettings):
    url: str = Field(validation_alias='REDIS_URL')
    decode_responses: bool = Field(validation_alias='REDIS_DECODE_RESPONSES')

@dataclass
class Redis:
    config: RedisConfig
    client: RedisClient = field(init=False)

    async def __connect__(self):
        self.client = await from_url(self.config.url, decode_responses=self.config.decode_responses)
        await self.client.ping()

    async def __disconnect__(self):
        await self.client.close()

    @property
    def db(self) -> RedisClient:
        return self.client

injector = DependencyInjector()
redis = injector.inject(Redis)()

async with injector:
    await redis.db.ping()
```

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

### Python: Dependency Injection with DependencyInjector

Source: https://context7.com/woltapp/magic-di/llms.txt

Illustrates the core usage of `DependencyInjector` for managing dependencies and their lifecycle. It shows how to define `Connectable` services, inject them into other classes, and use the injector as an asynchronous context manager to handle connection establishment and teardown. Dependencies are automatically discovered and managed.

```python
import asyncio
from magic_di import DependencyInjector, Connectable

class Database(Connectable):
    connected: bool = False

    async def __connect__(self) -> None:
        self.connected = True
        print("Database connected")

    async def __disconnect__(self) -> None:
        self.connected = False
        print("Database disconnected")

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

    def get_user(self, user_id: int) -> dict:
        return {"id": user_id, "connected": self.db.connected}

async def main():
    injector = DependencyInjector()

    # Inject dependencies - returns a factory function
    service_factory = injector.inject(UserService)
    service = service_factory()

    # Use async context manager for connection lifecycle
    async with injector:
        user = service.get_user(123)
        print(user)  # {"id": 123, "connected": True}

asyncio.run(main())
```

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

### Run Injected Functions with Lifecycle Management in Python

Source: https://context7.com/woltapp/magic-di/llms.txt

Introduces the `inject_and_run()` utility function, which simplifies the execution of asynchronous or synchronous functions. It automatically handles dependency injection, connection management (`__connect__`, `__disconnect__`), and execution.

```python
from magic_di import Connectable
from magic_di.utils import inject_and_run

class Database(Connectable):
    async def __connect__(self) -> None:
        print("Database connected")

    async def __disconnect__(self) -> None:
        print("Database disconnected")

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

    async def run(self) -> str:
        return "Work completed with database"

async def main(worker: Worker) -> str:
    result = await worker.run()
    print(result)
    return result

if __name__ == '__main__':
    # Automatically injects dependencies, connects, runs, and disconnects
    result = inject_and_run(main)
    # Output:
    # Database connected
    # Work completed with database
    # Database disconnected
```

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

### Integrate Magic-DI with FastAPI using inject_app()

Source: https://context7.com/woltapp/magic-di/llms.txt

Shows how to integrate Magic-DI with a FastAPI application using the `inject_app()` function. This enables automatic management of the connection lifecycle for `Connectable` services during application startup and shutdown.

```python
from fastapi import FastAPI
from magic_di import Connectable
from magic_di.fastapi import inject_app, Provide

app = inject_app(FastAPI())

class Database(Connectable):
    connected: bool = False

    async def __connect__(self) -> None:
        self.connected = True

    async def __disconnect__(self) -> None:
        self.connected = False

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

    def get_status(self) -> dict:
        return {"db_connected": self.db.connected}

@app.get("/status")
def get_status(service: Provide[UserService]) -> dict:
    return service.get_status()

@app.get("/users/{user_id}")
def get_user(user_id: int, service: Provide[UserService]) -> dict:
    return {"id": user_id, "status": service.get_status()}

# Run with: uvicorn app:app --reload
# GET /status -> {"db_connected": true}
```

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

### inject_and_run() Utility Function

Source: https://context7.com/woltapp/magic-di/llms.txt

Conveniently run injected async or sync functions with automatic connection lifecycle management.

```APIDOC
## inject_and_run() Utility Function

### Description
The `inject_and_run()` function provides a convenient way to run injected async or sync functions with automatic connection lifecycle management.

### Usage
```python
from magic_di import Connectable
from magic_di.utils import inject_and_run

class Database(Connectable):
    async def __connect__(self) -> None:
        print("Database connected")

    async def __disconnect__(self) -> None:
        print("Database disconnected")

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

    async def run(self) -> str:
        return "Work completed with database"

async def main(worker: Worker) -> str:
    result = await worker.run()
    print(result)
    return result

if __name__ == '__main__':
    # Automatically injects dependencies, connects, runs, and disconnects
    result = inject_and_run(main)
    # Output:
    # Database connected
    # Work completed with database
    # Database disconnected
```
```

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

### Use Custom Injector with FastAPI and Magic-DI

Source: https://context7.com/woltapp/magic-di/llms.txt

Demonstrates how to provide a custom `DependencyInjector` instance with pre-configured bindings to the FastAPI integration. This allows for more control over dependency resolution within the web application.

```python
from fastapi import FastAPI
from typing import Protocol
from magic_di import DependencyInjector, Connectable
from magic_di.fastapi import inject_app, Provide

class StorageInterface(Protocol):
    def save(self, data: bytes) -> str: ...

class S3Storage(Connectable):
    async def __connect__(self) -> None:
        print("S3 client initialized")

    async def __disconnect__(self) -> None:
        print("S3 client closed")

    def save(self, data: bytes) -> str:
        return "s3://bucket/file.bin"

# Create injector with bindings
injector = DependencyInjector()
injector.bind({StorageInterface: S3Storage})
```

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

### Bind Interfaces to Implementations in FastAPI

Source: https://github.com/woltapp/magic-di/blob/main/README.md

Explains how to use Protocols to define interfaces and bind them to specific implementations using the DI container for FastAPI endpoints.

```python
from typing import Protocol
from fastapi import FastAPI
from magic_di import Connectable, DependencyInjector
from magic_di.fastapi import inject_app, Provide

class MyInterface(Protocol):
    def do_something(self) -> bool:
        ...

class MyInterfaceImplementation(Connectable):
    def do_something(self) -> bool:
        return True

app = inject_app(FastAPI())
injector = DependencyInjector()
injector.bind({MyInterface: MyInterfaceImplementation})

@app.get(path="/hello-world")
def hello_world(service: Provide[MyInterface]) -> dict:
    return {"result": service.do_something()}
```

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

### Force Injection of Non-Connectable Dependencies

Source: https://github.com/woltapp/magic-di/blob/main/README.md

Shows how to use the Injectable type hint annotation to force the injection of dependencies that do not implement the Connectable protocol.

```python
from typing import Annotated
from magic_di import Injectable, Connectable

class Service(Connectable):
    dependency: Annotated[NonConnectableDependency, Injectable]
```

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

### FastAPI Integration: inject_app() Function

Source: https://context7.com/woltapp/magic-di/llms.txt

Integrates magic-di with FastAPI, automatically handling connection lifecycle during app startup and shutdown.

```APIDOC
## FastAPI Integration: inject_app() Function

### Description
The `inject_app()` function integrates magic-di with FastAPI, automatically handling connection lifecycle during app startup and shutdown.

### Usage
```python
from fastapi import FastAPI
from magic_di import Connectable
from magic_di.fastapi import inject_app, Provide

app = inject_app(FastAPI())

class Database(Connectable):
    connected: bool = False

    async def __connect__(self) -> None:
        self.connected = True

    async def __disconnect__(self) -> None:
        self.connected = False

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

    def get_status(self) -> dict:
        return {"db_connected": self.db.connected}

@app.get("/status")
def get_status(service: Provide[UserService]) -> dict:
    return service.get_status()

@app.get("/users/{user_id}")
def get_user(user_id: int, service: Provide[UserService]) -> dict:
    return {"id": user_id, "status": service.get_status()}

# Run with: uvicorn app:app --reload
# GET /status -> {"db_connected": true}
```
```

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

### FastAPI Dependency Injection with Custom Injector

Source: https://context7.com/woltapp/magic-di/llms.txt

Demonstrates how to integrate Magic-DI with FastAPI by passing a custom injector to the FastAPI application instance. This allows for dependency injection within FastAPI routes.

```python
from fastapi import FastAPI
from magic_di import inject_app, Provide

# Assuming 'injector' is a pre-configured Magic-DI injector instance
# injector = ... 
app = inject_app(FastAPI(), injector=injector)

class StorageInterface:
    def save(self, data: bytes) -> str:
        pass

class FileService(Connectable):
    def __init__(self, storage: StorageInterface):
        self.storage = storage

    def upload(self, data: bytes) -> str:
        return self.storage.save(data)

@app.post("/upload")
def upload_file(service: Provide[FileService]) -> dict:
    url = service.upload(b"file content")
    return {"url": url}
```

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

### Testing with InjectableMock in FastAPI

Source: https://context7.com/woltapp/magic-di/llms.txt

Demonstrates how to use InjectableMock with FastAPI and pytest to mock dependencies like EmailService. It shows overriding dependencies using injector.override() and verifying mock interactions.

```python
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from magic_di import DependencyInjector, Connectable
from magic_di.fastapi import inject_app, Provide
from magic_di.testing import InjectableMock, get_injectable_mock_cls

class EmailService(Connectable):
    async def __connect__(self) -> None:
        pass

    async def __disconnect__(self) -> None:
        pass

    async def send_email(self, to: str, subject: str, body: str) -> bool:
        # Real implementation
        return True

class NotificationService(Connectable):
    def __init__(self, email: EmailService):
        self.email = email

    async def notify(self, user_email: str, message: str) -> bool:
        return await self.email.send_email(user_email, "Notification", message)

app = inject_app(FastAPI())

@app.post("/notify")
async def send_notification(
    email: str,
    message: str,
    service: Provide[NotificationService]
) -> dict:
    success = await service.notify(email, message)
    return {"sent": success}

# Tests
@pytest.fixture
def email_mock():
    mock = InjectableMock()
    mock.send_email.return_value = True
    return mock

@pytest.fixture
def client(email_mock):
    injector = app.state.dependency_injector
    with injector.override({EmailService: email_mock.mock_cls}):
        with TestClient(app) as client:
            yield client

def test_notification_sent(client, email_mock):
    response = client.post("/notify?email=user@example.com&message=Hello")

    assert response.status_code == 200
    assert response.json() == {"sent": True}
    email_mock.send_email.assert_called_once_with(
        "user@example.com", "Notification", "Hello"
    )

def test_with_custom_mock(client):
    # Using get_injectable_mock_cls for custom mock objects
    class CustomEmailMock:
        async def send_email(self, to: str, subject: str, body: str) -> bool:
            return to.endswith("@valid.com")

    injector = app.state.dependency_injector
    with injector.override({EmailService: get_injectable_mock_cls(CustomEmailMock())}):
        with TestClient(app) as test_client:
            response = test_client.post("/notify?email=user@valid.com&message=Hi")
            assert response.json() == {"sent": True}

```

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

### Inject Non-Connectable Dependencies with Annotated in Python

Source: https://context7.com/woltapp/magic-di/llms.txt

Explains how to use the `Injectable` annotation with `typing.Annotated` to inject dependencies that do not implement the `Connectable` protocol. This is useful for configuration objects or other simple classes.

```python
from typing import Annotated
from magic_di import Injectable, Connectable, DependencyInjector

class Config:
    """A simple config class without __connect__/__disconnect__"""
    api_key: str = "secret-key"
    timeout: int = 30

class ApiClient(Connectable):
    def __init__(self, config: Annotated[Config, Injectable]):
        self.config = config

    async def __connect__(self) -> None:
        print(f"Connecting with timeout: {self.config.timeout}")

    async def __disconnect__(self) -> None:
        print("Disconnected")

injector = DependencyInjector()
client = injector.inject(ApiClient)()
# client.config is automatically injected despite Config not being Connectable
```

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

### Implement Healthchecks with Pingable Protocol

Source: https://github.com/woltapp/magic-di/blob/main/README.md

Explains how to implement the Pingable protocol to enable healthchecks. The DependenciesHealthcheck class automatically calls the __ping__ method on all injected clients that support this protocol.

```python
from magic_di.healthcheck import DependenciesHealthcheck

class Service(Connectable):
    def __init__(self, db: Database):
        self.db = db
    def is_connected(self):
        return self.db.connected
    async def __ping__(self) -> None:
        if not self.is_connected():
            raise Exception("Service is not connected")

@app.get(path="/healthcheck")
async def healthcheck_handler(healthcheck: Provide[DependenciesHealthcheck]) -> dict:
    await healthcheck.ping_dependencies()
    return {"alive": True}
```

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

### Integrate Dependency Injection with Celery Tasks

Source: https://github.com/woltapp/magic-di/blob/main/README.md

Demonstrates how to use InjectableCeleryTask to inject dependencies into both function-based and class-based Celery tasks.

```python
from celery import Celery
from magic_di.celery import get_celery_loader, InjectableCeleryTask, PROVIDE

app = Celery(loader=get_celery_loader(), task_cls=InjectableCeleryTask)

@app.task
async def calculate(x: int, y: int, calculator: Calculator = PROVIDE):
    await calculator.calculate(x, y)
```

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

### DependencyInjector.override() Context Manager

Source: https://context7.com/woltapp/magic-di/llms.txt

Temporarily replace bindings for testing or specialized scenarios. Original bindings are restored when the context exits.

```APIDOC
## DependencyInjector.override() Context Manager

### Description
Temporarily replace bindings for testing or specialized scenarios. Original bindings are restored when the context exits.

### Usage
```python
from magic_di import DependencyInjector, Connectable
from magic_di.testing import InjectableMock

class EmailService(Connectable):
    async def send_email(self, to: str, subject: str) -> bool:
        # Real email sending logic
        return True

class NotificationService(Connectable):
    def __init__(self, email: EmailService):
        self.email = email

    async def notify_user(self, user_email: str) -> bool:
        return await self.email.send_email(user_email, "Notification")

# In tests, override with mock
injector = DependencyInjector()

mock_email = InjectableMock()
mock_email.send_email.return_value = True

with injector.override({EmailService: mock_email.mock_cls}):
    service = injector.inject(NotificationService)()
    # service.email is now the mock
```
```

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

### FastAPI Integration: Custom Injector

Source: https://context7.com/woltapp/magic-di/llms.txt

Provide a custom injector with pre-configured bindings to the FastAPI integration.

```APIDOC
## FastAPI Integration: Custom Injector

### Description
You can provide a custom injector with pre-configured bindings to the FastAPI integration.

### Usage
```python
from fastapi import FastAPI
from typing import Protocol
from magic_di import DependencyInjector, Connectable
from magic_di.fastapi import inject_app, Provide

class StorageInterface(Protocol):
    def save(self, data: bytes) -> str: ...

class S3Storage(Connectable):
    async def __connect__(self) -> None:
        print("S3 client initialized")

    async def __disconnect__(self) -> None:
        print("S3 client closed")

    def save(self, data: bytes) -> str:
        return "s3://bucket/file.bin"

# Create injector with bindings
injector = DependencyInjector()
injector.bind({StorageInterface: S3Storage})

# You would then pass this injector to inject_app if it supported it, 
# or configure it globally before app startup.
# Example (conceptual, actual implementation might vary):
# app = inject_app(FastAPI(), injector=injector)
```
```

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

### Injectable Annotation

Source: https://context7.com/woltapp/magic-di/llms.txt

Use the `Injectable` annotation to force injection of non-connectable dependencies.

```APIDOC
## Injectable Annotation

### Description
Use the `Injectable` annotation to force injection of non-connectable dependencies. This is useful when you need to inject classes that don't implement the `Connectable` protocol.

### Usage
```python
from typing import Annotated
from magic_di import Injectable, Connectable, DependencyInjector

class Config:
    """A simple config class without __connect__/__disconnect__"""
    api_key: str = "secret-key"
    timeout: int = 30

class ApiClient(Connectable):
    def __init__(self, config: Annotated[Config, Injectable]):
        self.config = config

    async def __connect__(self) -> None:
        print(f"Connecting with timeout: {self.config.timeout}")

    async def __disconnect__(self) -> None:
        print("Disconnected")

injector = DependencyInjector()
client = injector.inject(ApiClient)()
# client.config is automatically injected despite Config not being Connectable
```
```

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

### FastAPI Healthcheck with Dependency Pinging

Source: https://context7.com/woltapp/magic-di/llms.txt

Implements a healthcheck endpoint in FastAPI using Magic-DI's `DependenciesHealthcheck`. This component automatically pings all registered dependencies that implement the `PingableProtocol` to verify their operational status.

```python
from fastapi import FastAPI, HTTPException
from magic_di import Connectable
from magic_di.fastapi import inject_app, Provide
from magic_di.healthcheck import DependenciesHealthcheck

app = inject_app(FastAPI())

class Database(Connectable):
    _connected: bool = False

    async def __connect__(self) -> None:
        self._connected = True

    async def __disconnect__(self) -> None:
        self._connected = False

    async def __ping__(self) -> None:
        if not self._connected:
            raise Exception("Database is not connected")
        print("Database ping successful")

class RedisCache(Connectable):
    async def __connect__(self) -> None:
        pass

    async def __disconnect__(self) -> None:
        pass

    async def __ping__(self) -> None:
        # Verify Redis connection
        print("Redis ping successful")

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

@app.get("/health")
async def healthcheck(health: Provide[DependenciesHealthcheck]) -> dict:
    try:
        await health.ping_dependencies(max_concurrency=2)
        return {"status": "healthy"}
    except Exception as e:
        raise HTTPException(status_code=503, detail=str(e))
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.