### Picodi Quick Start Example

Source: https://github.com/yakimka/picodi/blob/main/README.md

Demonstrates basic usage of Picodi for dependency injection in Python. It sets up configuration, injects a singleton httpx.AsyncClient, and retrieves data from the NASA API. It requires Python 3.10+ and the 'httpx' library.

```python
import asyncio
from collections.abc import Callable
from datetime import date
from typing import Any

import httpx

from picodi import (
    Provide,
    inject,
    registry,
    SingletonScope,
)
from picodi.helpers import get_value


# Regular functions without required arguments can be used as a dependency
def get_settings() -> dict:
    return {
        "nasa_api": {
            "api_key": "DEMO_KEY",
            "base_url": "https://api.nasa.gov",
            "timeout": 10,
        }
    }


# Helper function to get a setting from the settings dictionary.
# We can use this function to inject specific settings, not the whole settings object.
@inject
def get_setting(path: str, settings: dict = Provide(get_settings)) -> Callable[[], Any]:
    value = get_value(path, settings)
    return lambda: value


# We want to reuse the same client for all requests, so we declare a dependency
#   with `SingletonScope` that provides an httpx.AsyncClient instance
#   with the correct settings.
@registry.set_scope(
    scope_class=SingletonScope,
    auto_init=True,
)
@inject
async def get_nasa_client(
    api_key: str = Provide(get_setting("nasa_api.api_key")),
    base_url: str = Provide(get_setting("nasa_api.base_url")),
    timeout: int = Provide(get_setting("nasa_api.timeout")),
) -> httpx.AsyncClient:
    async with httpx.AsyncClient(
        base_url=base_url, params={"api_key": api_key}, timeout=timeout
    ) as client:
        yield client


@inject
async def get_apod(
    date: date, client: httpx.AsyncClient = Provide(get_nasa_client)
) -> dict[str, Any]:
    # Printing the client ID to show that the same client is reused for all requests.
    print("Client ID:", id(client))
    response = await client.get("/planetary/apod", params={"date": date.isoformat()})
    response.raise_for_status()
    return response.json()


@inject
# Note that asynchronous `get_nasa_client` is injected
#  in synchronous `print_client_info` function.
def print_client_info(client: httpx.AsyncClient = Provide(get_nasa_client)):
    print("Client ID:", id(client))
    print("Client Base URL:", client.base_url)
    print("Client Params:", client.params)
    print("Client Timeout:", client.timeout)


# `lifespan` will initialize dependencies on the application startup.
#   This will create the
#   httpx.AsyncClient instance and cache it for later use. Thereby, the same
#   client will be reused for all requests. This is important for connection
#   pooling and performance. Because it's created on app startup,
#   it will allow to pass asynchronous `get_nasa_client` into synchronous functions.
# And closing all inited dependencies after the function execution.
@registry.alifespan()
async def main():
    print_client_info()

    apod_data = await get_apod(date(2011, 7, 19))
    print("Title:", apod_data["title"])

    apod_data = await get_apod(date(2011, 7, 26))
    print("Title:", apod_data["title"])


asyncio.run(main())
# Client ID: 4334576784

```

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

### Async Yield Dependency with Async Context Manager in Python

Source: https://github.com/yakimka/picodi/blob/main/docs/tutorial/04_async_dependencies.rst

Demonstrates using `yield` with an asynchronous dependency provider, similar to `contextlib.asynccontextmanager`. This example uses a custom async context manager `AsyncDbConnection` to simulate database connection setup and teardown. Requires `asyncio` and `picodi`.

```python
# dependencies.py
import asyncio


# Assume this is an async context manager for a DB connection pool
class AsyncDbConnection:
    async def __aenter__(self):
        print("Async Yield Dep: Connecting to DB...")
        await asyncio.sleep(0.05)
        print("Async Yield Dep: Connected.")
        return self

    async def __aexit__(self, exc_type, exc, tb):
        print("Async Yield Dep: Disconnecting from DB...")
        await asyncio.sleep(0.05)
        print("Async Yield Dep: Disconnected.")

    async def execute(self, query: str):
        print(f"Async Yield Dep: Executing query '{query}'")
        await asyncio.sleep(0.02)
        return "Query Result"


async def get_db_connection():
    """Provides an async DB connection and ensures disconnection."""
    async with AsyncDbConnection() as connection:
        yield connection
```

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

### Install Picodi using pip

Source: https://github.com/yakimka/picodi/blob/main/README.md

This command installs the picodi library using the Python package installer (pip). It has no external dependencies.

```bash
pip install picodi
```

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

### Asynchronous Initialization Example - Python

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/lifespan.rst

Demonstrates how to handle asynchronous dependencies during initialization. If dependencies are async, `registry.init()` must be awaited in an async context.

```python
import asyncio
from picodi import registry, SingletonScope


@registry.set_scope(SingletonScope, auto_init=True)
async def get_async_service_client():
    print("Initializing Async Client...")
    await asyncio.sleep(0.1)
    return "AsyncServiceClient"


async def startup():
    print("App Starting...")
    # Must await because get_async_service_client is async
    await registry.init()
    print("Async Dependencies Initialized.")


asyncio.run(startup())
```

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

### Python: Dependency Injection Example - Without DI

Source: https://github.com/yakimka/picodi/blob/main/docs/introduction/index.rst

Demonstrates a class structure where a dependency (DatabaseConnection) is created internally by the class that needs it (UserService). This leads to tight coupling and makes testing difficult.

```Python
class DatabaseConnection:
    def __init__(self, connection_string: str):
        self._connection = connect(connection_string)  # Assume connect exists

    def fetch_data(self, query: str):
        # ... fetch data using self._connection
        pass


class UserService:
    def __init__(self):
        # UserService creates its own dependency
        self.db_connection = DatabaseConnection("my_db_string")

    def get_user(self, user_id: int):
        # ... uses self.db_connection
        pass
```

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

### Python: Dependency Injection Example - With DI

Source: https://github.com/yakimka/picodi/blob/main/docs/introduction/index.rst

Illustrates the Dependency Injection pattern where a dependency (DatabaseConnection) is provided externally to the class that needs it (UserService). This promotes decoupling and improves testability.

```Python
class DatabaseConnection:
    # ... (same as before)
    def __init__(self, connection_string: str): ...  # (same as before)


class UserService:
    def __init__(self, db_connection: DatabaseConnection):
        # Dependency is passed in (injected)
        self.db_connection = db_connection

    def get_user(self, user_id: int):
        # ... uses self.db_connection
        pass


# Somewhere else in the application (e.g., the entry point)
db_conn = DatabaseConnection("my_db_string")
user_service = UserService(db_conn)  # Injection happens here
```

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

### Define a simple dependency in Python

Source: https://github.com/yakimka/picodi/blob/main/docs/tutorial/01_first_steps.rst

Demonstrates defining a basic dependency in Picodi. Dependencies are typically Python functions that return a value and do not require arguments. This example defines a function to provide an API base URL.

```python
# dependencies.py
def get_api_base_url() -> str:
    """Provides the base URL for an external API."""
    print("Creating API base URL dependency")
    return "https://api.example.com"
```

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

### Example pytest output with Picodi override

Source: https://github.com/yakimka/picodi/blob/main/docs/tutorial/06_testing.rst

This example shows the typical output when running pytest tests with a Picodi marker override active. It highlights how the test environment reflects the overridden service URL.

```text
test_services_pytest.py
Test: Calling service with marker override active.
Test Override: Providing TEST URL
Service: Calling API at: http://test.server.com/test/endpoint
Test: Service call returned.
Test: Test function finished.
.
```

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

### Define nested dependencies in Python with Picodi

Source: https://github.com/yakimka/picodi/blob/main/docs/tutorial/01_first_steps.rst

Demonstrates how to create a dependency that itself depends on another dependency. Picodi automatically resolves the entire chain of dependencies. This example shows a URL provider depending on a configuration provider.

```python
# dependencies.py
from picodi import Provide, inject


def get_config() -> dict:
    """Provides application configuration."""
    print("Loading config")
    return {"api_url": "https://api.config.com"}


@inject  # Inject config here
def get_api_base_url(config: dict = Provide(get_config)) -> str:
    """Provides the base URL from config."""
    print("Creating API base URL from config")
    return config["api_url"]
```

```python
# services.py
# (call_external_api remains the same, using get_api_base_url)
# from dependencies import get_api_base_url
from picodi import Provide, inject


@inject
def call_external_api(endpoint: str, base_url: str = Provide(get_api_base_url)) -> str:
    """Calls an endpoint on the external API."""
    full_url = f"{base_url}/{endpoint.lstrip('/')}"
    print(f"Calling API at: {full_url}")
    return f"Response from {full_url}"
```

```python
# main.py
# from services import call_external_api

response = call_external_api("orders")
print(response)
```

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

### Generator Dependencies with Lifecycle Management in Python

Source: https://context7.com/yakimka/picodi/llms.txt

Illustrates using generator functions as dependencies to manage resource setup and teardown automatically. This is ideal for resources like database connections or file handles, ensuring they are properly opened and closed. The example shows a generator for a database session.

```python
from picodi import inject, Provide

def get_db_session():
    print("Opening database connection")
    session = "DB Session"
    yield session
    print("Closing database connection")

@inject
def create_user(name: str, db: str = Provide(get_db_session)):
    print(f"Creating user {name} using {db}")
    return f"User {name} created"

result = create_user("Alice")
# Output:
# Opening database connection
# Creating user Alice using DB Session
# Closing database connection
```

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

### Registry Management and Dependency Overrides in Python

Source: https://context7.com/yakimka/picodi/llms.txt

Explains how to use the `registry` to override dependencies, which is valuable for testing or managing different execution contexts without modifying the original code. The example shows switching between a production database and a test database for a data querying function.

```python
from picodi import inject, Provide, registry

def get_database():
    return "Production Database"

def get_test_database():
    return "Test Database"

@inject
def query_data(db: str = Provide(get_database)):
    return f"Querying {db}"

# Production usage
result = query_data()
# Output: "Querying Production Database"

# Testing with override
with registry.override(get_database, get_test_database):
    result = query_data()
    # Output: "Querying Test Database"

# Back to production
result = query_data()
# Output: "Querying Production Database"
```

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

### Define Async Yield Dependency Provider - Python

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/async.rst

For asynchronous resources requiring setup and teardown, use an `async def` function with a single `yield`. This functions like `contextlib.asynccontextmanager`, allowing Picodi to manage the resource's lifecycle by awaiting setup before yield and teardown after yield.

```python
import asyncio


class AsyncDbClient:
    async def connect(self):
        print("Async Yield Dep: Connecting...")
        await asyncio.sleep(0.05)
        return self

    async def close(self):
        print("Async Yield Dep: Closing connection...")
        await asyncio.sleep(0.05)

    async def query(self, sql):
        print(f"Async Yield Dep: Running query: {sql}")
        await asyncio.sleep(0.1)
        return [{"id": 1}, {"id": 2}]


async def get_db_client():
    client = AsyncDbClient()
    await client.connect()
    try:
        yield client  # Yield the connected client
    finally:
        await client.close()
```

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

### Async Singleton Lifespan Management with Picodi

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/async.rst

Demonstrates how to manage the lifespan of an asynchronous singleton dependency using Picodi's registry. It shows the initialization, main logic execution, and cleanup phases, ensuring async setup and teardown are handled correctly.

```python
import asyncio
from picodi import registry

@registry.alifespan()
async def res():
    print("Async Singleton: Init")
    yield "Async Resource Data"
    print("Async Singleton: Cleanup")

async def main_logic():
    print(f"Main logic using: {registry.get(res)}")


async def run():
    async with registry.alifespan():  # Handles await init() and await shutdown()
        await main_logic()


asyncio.run(run())
```

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

### Manage Resources with Yield Dependencies (Synchronous) in Python

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/dependencies.rst

Shows how to use generator functions with a single `yield` statement to manage resources like database connections. Picodi treats these as context managers, ensuring setup and teardown logic is executed correctly.

```python
import sqlite3


def get_db_cursor():
    """Provides a database cursor and ensures the connection is closed."""
    connection = sqlite3.connect(":memory:")
    print("DB Connection Opened")
    cursor = connection.cursor()
    try:
        yield cursor  # Provide the cursor
    finally:
        connection.close()
        print("DB Connection Closed")
```

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

### FastAPI Request-Scoped Dependencies Setup

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/integrations.rst

Shows the setup for managing request-scoped dependencies in FastAPI using Picodi's RequestScopeMiddleware and RequestScope. The RequestScopeMiddleware must be placed first in the middleware stack.

```python
import uuid

from fastapi import FastAPI
from picodi import registry
from picodi.integrations.fastapi import Provide, RequestScope, RequestScopeMiddleware
from starlette.middleware import Middleware


# Define request-scoped dependency

app = FastAPI(
    middleware=[
        Middleware(RequestScopeMiddleware), # Must be added first
        # ... other middlewares ...
    ]
)


# Example of using RequestScope (typically within a dependency function)
@registry.register_singleton(RequestScope)
def get_request_id():
    return str(uuid.uuid4())


@app.get("/request-scoped")
async def get_request_id_route(
    request_id: str = Provide(get_request_id)
):
    return {"request_id": request_id}
```

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

### Run Async Yield Service with `asyncio.run` in Python

Source: https://github.com/yakimka/picodi/blob/main/docs/tutorial/04_async_dependencies.rst

Illustrates running an asynchronous service that utilizes async yield dependencies using `asyncio.run()`. This example demonstrates executing a database query through an injected async connection. Requires `asyncio`.

```python
# main.py
import asyncio

# from services import run_db_query

print("Main: Running async DB service.")
result = asyncio.run(run_db_query("SELECT * FROM users"))
print(f"Main: Got result: {result}")
print("Main: Async DB service finished.")
```

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

### Manage Resources with Yield Dependencies (Asynchronous) in Python

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/dependencies.rst

Demonstrates using asynchronous generator functions with `yield` for managing asynchronous resources. This pattern ensures that asynchronous setup and teardown operations are handled properly.

```python
import asyncio


class AsyncResource:
    async def setup(self):
        print("Async Resource Setup")
        await asyncio.sleep(0.05)
        return self

    async def close(self):
        print("Async Resource Closed")
        await asyncio.sleep(0.05)

    async def do_work(self):
        print("Async Resource Working")


async def get_async_resource():
    """Provides an async resource with setup and teardown."""
    resource = AsyncResource()
    await resource.setup()
    try:
        yield resource
    finally:
        await resource.close()
```

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

### Picodi SingletonScope Example

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/scopes.rst

Demonstrates how to use SingletonScope to create a shared, single instance of a resource that persists until registry shutdown. This is useful for expensive objects like database connection pools.

```python
from picodi import registry, SingletonScope, Provide, inject


@registry.set_scope(SingletonScope)  # Assign SingletonScope here
def get_shared_resource():
    print("Creating shared resource...")
    # Imagine this is an expensive object like a DB connection pool
    resource = {"id": "singleton_resource"}
    try:
        yield resource
    finally:
        print("Cleaning up shared resource...")


@inject
def use_resource_1(res=Provide(get_shared_resource)):
    print(f"User 1 using resource: {res['id']}")


@inject
def use_resource_2(res=Provide(get_shared_resource)):
    print(f"User 2 using resource: {res['id']}")


# --- Application Code ---
print("First use:")
use_resource_1()
print("\nSecond use:")
use_resource_2()

print("\nShutting down:")
# SingletonScope requires manual shutdown for cleanup
registry.shutdown()
```

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

### Inject Dependencies at Application Edges (FastAPI Example)

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/best_practices.rst

This example demonstrates injecting dependencies at the application edge, specifically within a FastAPI route handler. Dependencies are provided to the service constructor without direct DI decoration on the service class itself, promoting decoupling. The `UserService` receives its dependencies (`Database`, `Cache`) through its constructor, and the `get_user_service` provider is decorated with `@inject` to resolve these dependencies using `Provide`. The FastAPI route then injects the `UserService` provider using `Provide(get_user_service, wrap=True)`.

```python
# --- Dependencies ---
class Database:
    ...


def get_db() -> Database:
    ...


class Cache:
    ...


def get_cache() -> Cache:
    ...


# --- Service Layer (NO injection here) ---
class UserService:
    # Receives dependencies via constructor, but NOT via ``@inject``
    def __init__(self, db: Database, cache: Cache):
        self.db = db
        self.cache = cache

    def get_user_profile(self, user_id: int):
        # Uses db and cache
        ...


# --- BETTER: Inject dependencies needed to *create* the service ---
from picodi import Provide, inject


@inject  # Inject DB and Cache into the *service provider*
def get_user_service(
    db: Database = Provide(get_db),
    cache: Cache = Provide(get_cache)
) -> UserService:
    # The service itself doesn't use ``@inject`` for its constructor
    return UserService(db=db, cache=cache)


# --- FastAPI Route (Injection happens HERE) ---
from fastapi import FastAPI
from picodi.integrations.fastapi import Provide  # Use FastAPI version

app = FastAPI()


@app.get("/users/{user_id}")
async def read_user(
    user_id: int,
    # Inject the UserService provider at the edge (route)
    user_service: UserService = Provide(get_user_service, wrap=True),
):
    profile = user_service.get_user_profile(user_id)
    return profile

```

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

### Async Shutdown Example - Python

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/lifespan.rst

An example demonstrating how to shut down the Picodi registry and clear it, typically used at the end of an application's lifecycle. It ensures that any asynchronous cleanup tasks are properly awaited.

```python
import asyncio
from picodi import registry


async def teardown():
    await registry.shutdown()
    registry._clear()


asyncio.run(teardown())
```

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

### Async Dependency Injection in Python

Source: https://context7.com/yakimka/picodi/llms.txt

Demonstrates Picodi's capability to handle asynchronous dependencies seamlessly. This allows async functions to be injected into both synchronous and asynchronous contexts, utilizing libraries like `httpx` for network requests. The example shows fetching user data using an async HTTP client.

```python
import asyncio
from picodi import inject, Provide, registry, SingletonScope
import httpx

@registry.set_scope(SingletonScope, auto_init=True)
async def get_http_client():
    async with httpx.AsyncClient(base_url="https://api.example.com") as client:
        yield client

@inject
async def fetch_user(user_id: int, client: httpx.AsyncClient = Provide(get_http_client)):
    response = await client.get(f"/users/{user_id}")
    return response.json()

@registry.alifespan()
async def main():
    user_data = await fetch_user(123)
    print(f"User data: {user_data}")

asyncio.run(main())
```

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

### Starlette Request-Scoped Dependency Example

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/integrations.rst

Demonstrates how to define and use a request-scoped dependency in Starlette using Picodi. The RequestScope ensures that the dependency instance is unique per HTTP request and is automatically cleaned up. This example requires the `RequestScopeMiddleware` to be added to the Starlette application.

```python
# dependencies.py
from picodi import registry
from picodi.integrations.starlette import RequestScope
import uuid


@registry.set_scope(RequestScope)
def get_request_id():
    req_id = str(uuid.uuid4())
    print(f"REQUEST SCOPE: Generated request ID: {req_id}")
    yield req_id  # Use yield if cleanup needed, otherwise return
    print(f"REQUEST SCOPE: Cleaning up request ID: {req_id}")


# app.py
from picodi import Provide, inject
from picodi.integrations.starlette import RequestScopeMiddleware
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

# from dependencies import get_request_id


@inject
def service(request_id: str = Provide(get_request_id)):
    return request_id


@inject
async def homepage(_: Request, request_id: str = Provide(get_request_id)):
    # The request_id will be unique per request
    request_id_from_service = service()
    return JSONResponse(
        {
            "request_id": request_id,
            "request_id_from_service": request_id_from_service,
        }
    )


routes = [
    Route("/", homepage),
]

# Add the middleware
middleware = [
    Middleware(RequestScopeMiddleware)
    # You can optionally pass dependencies_for_init to the middleware
    # Middleware(RequestScopeMiddleware, dependencies_for_init=[dep1, dep2])
]

app = Starlette(routes=routes, middleware=middleware)

# Run with: uvicorn app:app
# Accessing '/' will show the same request_id from the service and the view
# this is because `get_request_id` has `RequestScope` scope

```

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

### Basic Dependency Injection with @inject in Python

Source: https://context7.com/yakimka/picodi/llms.txt

Demonstrates how to use the @inject decorator to automatically provide dependencies to functions. It replaces `Provide` markers with resolved dependencies, simplifying function calls. This example shows injecting a database connection into a user fetching function.

```python
from picodi import inject, Provide

def get_database_connection():
    return "PostgreSQL Connection"

@inject
def fetch_user(user_id: int, db: str = Provide(get_database_connection)):
    return f"Fetching user {user_id} from {db}"

result = fetch_user(42)
# Output: "Fetching user 42 from PostgreSQL Connection"
```

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

### Inject Dependencies into Methods using Picodi

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/injection.rst

Demonstrates how to use the @inject decorator and Provide to inject dependencies into class methods, including the __init__ method. This example shows dependency injection for a logger service.

```python
from picodi import Provide, inject


def get_logger():
    print("Creating logger")
    return "MyLogger"


class MyService:
    @inject
    def __init__(self, logger=Provide(get_logger)):
        print("MyService.__init__ called")
        self.logger = logger

    def do_something(self):
        print(f"Doing something with {self.logger}")


service = MyService()
service.do_something()
```

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

### Pytest Integration for Picodi Dependency Management

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/testing.rst

Shows how to integrate Picodi with Pytest by adding plugins to `conftest.py` for automatic setup and cleanup. It explains the automatic cleanup actions performed after each test, including shutting down manual-scoped dependencies and clearing overrides.

```python
# conftest.py
pytest_plugins = [
    "picodi.integrations._pytest",
    # If you use asyncio in your tests, add the following plugin as well.
    # It must be added after the main plugin.
    # "picodi.integrations._pytest_asyncio",
]
```

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

### Python: Manage Temp File Lifecycle with Yield Dependency

Source: https://github.com/yakimka/picodi/blob/main/docs/tutorial/02_yield_dependencies.rst

Demonstrates managing a temporary file's creation and deletion using a Python generator function with a single yield. The setup phase creates the file, yields its path for use by injecting functions, and the teardown phase (in a finally block) removes the file after use. This pattern ensures resources are properly managed.

```python
# dependencies.py
import tempfile
import os


# We don't need @inject or Provide here for get_temp_file_path
# because it doesn't depend on any other dependencies.
def get_temp_file_path():
    """Provides a path to a temporary file and cleans it up afterwards."""
    tf = tempfile.NamedTemporaryFile(delete=False, mode="w+", suffix=".txt")
    file_path = tf.name
    print("Setup: Created temp file")
    tf.close()  # Close the file handle, but the file remains

    try:
        yield file_path  # Provide the path
    finally:
        # Teardown: This code runs after the injecting function finishes
        if os.path.exists(file_path):
            os.remove(file_path)
            print("Teardown: Removed temp file")
        else:
            print(
                "Teardown: Temp file already removed"
            )  # Should not happen in normal flow


# services.py
from picodi import Provide, inject


# from dependencies import get_temp_file_path


@inject
def write_to_temp_file(
    content: str,
    temp_file: str = Provide(get_temp_file_path),  # Inject the yielded path
) -> None:
    """Writes content to a temporary file provided by a dependency."""
    print("Service: Writing to temp_file")
    with open(temp_file, "a") as f:
        f.write(content + "\n")
    print("Service: Finished writing to temp_file")


# main.py
# from services import write_to_temp_file

print("Main: Calling service the first time.")
write_to_temp_file("Hello from Picodi!")
print("Main: Service call finished.")

print("\nMain: Calling service the second time.")
write_to_temp_file("Another message.")
print("Main: Service call finished.")

```

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

### Eager Singleton Initialization with Picodi

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/scopes.rst

Demonstrates how to configure a singleton dependency to be eagerly initialized when the application starts using `auto_init=True`. This ensures the dependency is created upon calling `registry.init()`, rather than on its first injection. It uses picodi's registry, inject, and Provide decorators.

```python
from picodi import inject, registry, Provide, SingletonScope


@registry.set_scope(SingletonScope, auto_init=True)  # Note auto_init
def get_eager_singleton():
    print("Eager singleton created!")
    return "I was created early"


# At application startup:
print("Calling registry.init()...")
registry.init()  # This will initialize all 'auto_init=True' dependencies
print("registry.init() finished.")


# Later, when injected:
@inject
def use_eager(dep=Provide(get_eager_singleton)):
    print(f"Using dependency: {dep}")


use_eager()  # Will not print "Eager singleton created!" again
```

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

### Configure Picodi Pytest Plugin in conftest.py

Source: https://github.com/yakimka/picodi/blob/main/docs/tutorial/06_testing.rst

This Python snippet shows how to integrate Picodi with Pytest by adding the necessary plugin to your `conftest.py` file. This setup enables automatic cleanup of Picodi's registry state after each test, including shutting down scoped dependencies and clearing overrides, ensuring test isolation.

```python
# conftest.py
pytest_plugins = [
    "picodi.integrations._pytest",
    # If you use asyncio in your tests, add the following plugin as well.
    # It must be added after the main plugin.
    # "picodi.integrations._pytest_asyncio",
]

```

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

### Singleton Scope for Shared Resources in Python

Source: https://context7.com/yakimka/picodi/llms.txt

Shows how to configure dependencies with `SingletonScope` to ensure a single instance is created and reused throughout the application's lifecycle. This is useful for expensive resources like API clients. The example demonstrates reusing an API client across multiple calls.

```python
from picodi import inject, Provide, registry, SingletonScope

@registry.set_scope(SingletonScope, auto_init=True)
def get_api_client():
    print("Creating expensive API client")
    client = {"client_id": "ABC123"}
    yield client
    print("Cleaning up API client")

@inject
def fetch_data(endpoint: str, client: dict = Provide(get_api_client)):
    print(f"Client ID: {client['client_id']}, Endpoint: {endpoint}")
    return f"Data from {endpoint}"

@registry.alifespan()
async def main():
    result1 = fetch_data("/users")
    result2 = fetch_data("/posts")
    # Same client instance reused

import asyncio
asyncio.run(main())
# Output:
# Creating expensive API client
# Client ID: ABC123, Endpoint: /users
# Client ID: ABC123, Endpoint: /posts
# Cleaning up API client
```

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

### Inject Async Dependencies into Async Function - Python

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/async.rst

If a function needs to inject an asynchronous dependency, the function itself must be `async def`. Picodi awaits the asynchronous dependency provider during injection. This example demonstrates injecting both async fetch and async yield dependencies.

```python
import asyncio
from picodi import Provide, inject

# Assume async dependencies from above are defined


@inject
async def process_data(
    config: dict = Provide(fetch_remote_config),
    db_client=Provide(get_db_client),  # Injecting async yield dep
):
    print(f"Async Service: Got config: {config}")
    if config.get("feature_x_enabled"):
        results = await db_client.query("SELECT * FROM data")
        print(f"Async Service: Got DB results: {results}")


asyncio.run(process_data())
```

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

### Initialize Dependency with Type Hints - Python

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/best_practices.rst

Demonstrates how to define a dependency provider function with clear Python type hints for the return type and injected parameters. This enhances readability and enables static analysis tools like MyPy to verify type correctness.

```python
from picodi import Provide, inject


class MyClient: ...


# Good: Clear type hints
def get_my_client() -> MyClient:
    return MyClient()


@inject
def use_the_client(client: MyClient = Provide(get_my_client)):
    # Mypy can verify 'client' is used correctly
    pass
```

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

### GET /request-id

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/integrations.rst

This endpoint retrieves a unique correlation ID generated for the current request using Picodi's request scope.

```APIDOC
## GET /request-id

### Description
Retrieves the correlation ID generated for the current request. This ID is managed within the request scope by Picodi middleware.

### Method
GET

### Endpoint
/request-id

### Parameters
#### Query Parameters
None

#### Path Parameters
None

#### Request Body
None

### Request Example
None

### Response
#### Success Response (200)
- **correlation_id** (string) - The unique correlation ID for the request.

#### Response Example
{
  "correlation_id": "a1b2c3d4"
}
```

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

### Synchronous and Asynchronous Dependency Resolution with Picodi

Source: https://context7.com/yakimka/picodi/llms.txt

Demonstrates how to resolve dependencies synchronously using a context manager and asynchronously using an async context manager. This pattern is useful for managing configurations or services that might involve I/O operations.

```python
with registry.resolve(get_config) as config:
    print(f"API Key: {config['api_key']}")
    print(f"Timeout: {config['timeout']}")

import asyncio

async def get_async_config():
    await asyncio.sleep(0.1)
    return {"service": "API", "version": "v2"}

async def main():
    async with registry.aresolve(get_async_config) as config:
        print(f"Service: {config['service']}, Version: {config['version']}")

asyncio.run(main())
```

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

### FastAPI Route with Provide(wrap=True)

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/integrations.rst

Presents the recommended method for injecting Picodi dependencies into FastAPI routes using Provide(wrap=True). This approach avoids explicit @inject on the route and integrates directly with FastAPI's DI system. Assumes 'get_my_service' is a defined Picodi dependency.

```python
from fastapi import FastAPI
from picodi.integrations.fastapi import Provide  # Use the fastapi version

app = FastAPI()


# Assume get_my_service is defined as before
def get_my_service():
    print("Providing my_service")
    return "My Service Instance"


@app.get("/wrapped-route")
async def route_without_inject(
    # No @inject needed on the route!
    # Provide(..., wrap=True) integrates with FastAPI's DI
    service_instance: str = Provide(get_my_service, wrap=True)
):
    return {"service": service_instance}
```

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

### Run Async Service with `asyncio.run` in Python

Source: https://github.com/yakimka/picodi/blob/main/docs/tutorial/04_async_dependencies.rst

Illustrates the standard way to run an asynchronous main function or service in Python using `asyncio.run()`. This is necessary to execute the async code that involves Picodi's async dependencies. Requires `asyncio`.

```python
# main.py
import asyncio

# from services import process_user

print("Main: Running async service.")
asyncio.run(process_user())
print("Main: Async service finished.")
```

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

### Calling registry.init() - Python

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/lifespan.rst

Illustrates the basic usage of `registry.init()` for dependency initialization during application startup. This call triggers the creation of dependencies marked for initialization.

```python
# At application startup
print("App Starting...")
registry.init()
# If you have async dependencies marked for init, instead you MUST await
# await registry.init()
print("Dependencies Initialized.")

# Application runs...
```

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

### Lifecycle Management with Init and Shutdown in Python

Source: https://context7.com/yakimka/picodi/llms.txt

Shows explicit control over dependency initialization and cleanup in Python using Picodi's lifecycle management features. It demonstrates manual initialization and shutdown for resources like a cache.

```python
from picodi import registry, inject, Provide, SingletonScope

@registry.set_scope(SingletonScope)
def get_cache():
    print("Initializing cache")
    cache = {"storage": {}}
    yield cache
    print("Flushing cache")

@inject
def cache_user(user_id: int, cache: dict = Provide(get_cache)):
    cache["storage"][user_id] = f"User {user_id}"
    return f"Cached user {user_id}"

# Manual lifecycle control
async def main():
    # Initialize all auto_init dependencies
    await registry.init([get_cache])

    # Use dependencies
    result1 = cache_user(1)
    result2 = cache_user(2)
    print(result1, result2)

    # Cleanup when done
    await registry.shutdown(SingletonScope)

import asyncio
asyncio.run(main())
```

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

### Initialize Dependencies with auto_init=True - Python

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/lifespan.rst

Demonstrates how to register a dependency for automatic initialization when `registry.init()` is called. This is useful for singletons that should be ready at application startup.

```python
from picodi import registry, SingletonScope


@registry.set_scope(SingletonScope, auto_init=True)
def get_cache_client():
    print("Initializing Cache Client...")
    # ... create and return client ...
    return "RedisClient"
```

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

### Define Simple Synchronous Dependency Providers in Python

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/dependencies.rst

Demonstrates defining basic dependency providers as synchronous Python functions that return values. These functions are typically used to provide configuration settings or data access objects.

```python
def get_database_url() -> str:
    """Returns the connection string for the database."""
    return "postgresql://user:password@host:port/dbname"

def get_settings() -> dict:
    """Loads and returns application settings."""
    # In a real app, this might load from a file or environment variables
    return {"timeout": 30, "retries": 3}
```

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

### Applying multiple dependency overrides with pytest marker

Source: https://github.com/yakimka/picodi/blob/main/docs/tutorial/06_testing.rst

Demonstrates how to apply multiple dependency overrides simultaneously by passing a list of tuples to the `@pytest.mark.picodi_override` marker. This allows for more complex test configurations.

```python
@pytest.mark.picodi_override([(dep1, override1), (dep2, override2)])
```

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

### Define Async Dependency with `async def` in Python

Source: https://github.com/yakimka/picodi/blob/main/docs/tutorial/04_async_dependencies.rst

Demonstrates defining an asynchronous dependency provider using `async def`. This function simulates an I/O-bound operation like fetching data from a network service. It requires the `asyncio` library.

```python
# dependencies.py
import asyncio


async def fetch_data() -> dict:
    """Simulates fetching data asynchronously."""
    print("Async Dep: Starting fetch")
    await asyncio.sleep(0.1)  # Simulate network delay
    print("Async Dep: Finished fetch")
    return {"data": "Data"}
```

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

### Explicit Ad-hoc Initialization - Python

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/lifespan.rst

Shows how to initialize specific dependencies on demand by passing a list of dependencies directly to `registry.init()`. This bypasses the auto-initialization and `add_for_init` configurations.

```python
registry.init([my_specific_dep_1, my_specific_dep_2])
```

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

### FastAPI Integration with Picodi RequestScope

Source: https://context7.com/yakimka/picodi/llms.txt

Demonstrates integrating Picodi with FastAPI using `RequestScope` and `RequestScopeMiddleware` to manage dependencies that are unique to each incoming request. It also shows an alternative using `wrap=True` with FastAPI's native dependency system.

```python
from fastapi import FastAPI
from picodi.integrations.fastapi import Provide, RequestScope, RequestScopeMiddleware
from picodi import inject, registry

app = FastAPI()
app.add_middleware(RequestScopeMiddleware)

@registry.set_scope(RequestScope)
def get_request_id():
    import uuid
    request_id = str(uuid.uuid4())
    print(f"Generated request ID: {request_id}")
    return request_id

@app.get("/api/users/{user_id}")
@inject
async def get_user(user_id: int, request_id: str = Provide(get_request_id)):
    return {
        "user_id": user_id,
        "request_id": request_id,
        "message": f"Fetched user {user_id}"
    }

# Alternative: Use FastAPI native dependency system with wrap=True
@app.get("/api/posts/{post_id}")
async def get_post(post_id: int, request_id: str = Provide(get_request_id, wrap=True)):
    return {
        "post_id": post_id,
        "request_id": request_id
    }
```

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

### Define Simple Asynchronous Dependency Providers in Python

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/dependencies.rst

Illustrates how to define asynchronous dependency providers using `async def`. These are useful for I/O-bound operations like fetching data from external APIs or interacting with asynchronous resources.

```python
import asyncio


async def get_external_api_key() -> str:
    """Fetches an API key from a secure vault (simulated)."""
    print("Fetching API key...")
    await asyncio.sleep(0.1)  # Simulate I/O
    return "secret-api-key-12345"
```

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

### Python: Asynchronous Lifespan Context Manager

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/lifespan.rst

Presents the `registry.alifespan()` asynchronous context manager for applications with an asynchronous main lifecycle. This context manager handles `await registry.init()` upon entry and `await registry.shutdown()` upon exit, ensuring proper asynchronous initialization and cleanup of dependencies, including async generators.

```python
import asyncio
from picodi import registry, SingletonScope, Provide, inject


@registry.set_scope(SingletonScope, auto_init=True)
async def get_async_singleton():
    print("Async Singleton Init")
    await asyncio.sleep(0.05)
    yield "Async Data"
    print("Async Singleton Cleanup")
    await asyncio.sleep(0.05)


@inject
async def main_async_logic(data=Provide(get_async_singleton)):
    print(f"Running async logic with: {data}")


async def run_app():
    print("Entering alifespan...")
    async with registry.alifespan():  # Handles await init() and await shutdown()
        await main_async_logic()
    print("Exited alifespan.")


asyncio.run(run_app())
```

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

### Use a function with injected dependencies in Python

Source: https://github.com/yakimka/picodi/blob/main/docs/tutorial/01_first_steps.rst

Illustrates how to call a function that has dependencies injected by Picodi. The function is called like a regular Python function, with Picodi handling the dependency resolution automatically.

```python
# main.py
# from services import call_external_api

response = call_external_api("users")
print(response)

response_2 = call_external_api("products")
print(response_2)
```

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

### Inject a dependency into a function using Picodi

Source: https://github.com/yakimka/picodi/blob/main/docs/tutorial/01_first_steps.rst

Shows how to inject a dependency into a Python function using Picodi's `@inject` decorator and `Provide` marker. This allows the function to automatically receive the dependency's value when called.

```python
# services.py
from picodi import Provide, inject

# Assume dependencies.py is in the same directory or Python path
# from dependencies import get_api_base_url


@inject
def call_external_api(
    endpoint: str, base_url: str = Provide(get_api_base_url)  # Inject here!
) -> str:
    """Calls an endpoint on the external API."""
    full_url = f"{base_url}/{endpoint.lstrip('/')}"
    print(f"Calling API at: {full_url}")
    # In a real app, you'd use an HTTP client here
    return f"Response from {full_url}"
```

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

### Async Lifespan Management with Registry - Python

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/async.rst

When using async dependencies with specific scopes or eager initialization, `registry.init()` and `registry.shutdown()` must be awaited as they return awaitables. The `alifespan` context manager automates this for async lifecycles.

```python
import asyncio
from picodi import registry, SingletonScope, Provide, inject


@registry.set_scope(SingletonScope, auto_init=True)
async def get_async_singleton_resource():
    print("Async Singleton: Init")
    yield "Async Resource Data"
    print("Async Singleton: Cleanup")


@inject
async def main_logic(res=Provide(get_async_singleton_resource)):
    print(f"Main logic using resource: {res}")

async def run_lifespan():
    async with registry.alifespan():
        await main_logic()

asyncio.run(run_lifespan())
```

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

### ContextVarScope for Context-Local Dependencies in Python

Source: https://context7.com/yakimka/picodi/llms.txt

Demonstrates how to use ContextVarScope to provide separate dependency instances per context, such as per task in concurrent Python applications. It initializes and cleans up contexts automatically.

```python
import asyncio
from picodi import inject, Provide, registry, ContextVarScope

@registry.set_scope(ContextVarScope)
def get_task_context():
    import threading
    context = {
        "task_id": id(asyncio.current_task()),
        "thread_id": threading.current_thread().ident
    }
    print(f"Created context: {context}")
    yield context
    print(f"Cleaning up context: {context}")

@inject
async def process_item(item: str, context: dict = Provide(get_task_context)):
    await asyncio.sleep(0.1)
    print(f"Processing {item} in task {context['task_id']}")
    return f"{item} processed"

async def main():
    # Each task gets its own context
    tasks = [process_item(f"item-{i}") for i in range(3)]
    results = await asyncio.gather(*tasks)

    # Cleanup all contexts
    await registry.shutdown(ContextVarScope)

asyncio.run(main())
```

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

### Initialize Dependencies with add_for_init - Python

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/lifespan.rst

Shows how to explicitly add dependencies to the initialization list using `registry.add_for_init()`. This method allows for precise control over which dependencies are initialized.

```python
from picodi import registry, SingletonScope


@registry.set_scope(SingletonScope)  # No auto_init here
def get_db_pool():
    print("Initializing DB Pool...")
    # ... create and return pool ...
    return "DbPool"


# Explicitly add it to the init list
registry.add_for_init([get_db_pool])  # Can pass a list or callable returning a list
```

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

### Python: Synchronous Lifespan Context Manager

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/lifespan.rst

Demonstrates the use of `registry.lifespan()` as a synchronous context manager for automatic dependency initialization and cleanup. This is ideal for applications with a synchronous main lifecycle, as it handles calling `registry.init()` upon entry and `registry.shutdown()` upon exit, simplifying resource management.

```python
from picodi import registry, SingletonScope, Provide, inject


@registry.set_scope(SingletonScope, auto_init=True)
def get_sync_singleton():
    print("Sync Singleton Init")
    yield "Sync Data"
    print("Sync Singleton Cleanup")


@inject
def main_sync_logic(data=Provide(get_sync_singleton)):
    print(f"Running sync logic with: {data}")


print("Entering lifespan...")
with registry.lifespan():  # Handles init() and shutdown()
    main_sync_logic()
print("Exited lifespan.")
```

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

### FastAPI Route Combining FastAPI and Picodi Dependencies

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/integrations.rst

Illustrates how to combine standard FastAPI dependencies (like path parameters and security) with Picodi-managed dependencies within a single route function signature. Uses Provide(wrap=True) for the Picodi dependency.

```python
from fastapi import FastAPI, Depends, Path, HTTPException
from picodi.integrations.fastapi import Provide
from typing import Annotated

app = FastAPI()


# --- Picodi Dependency ---
class DatabaseClient:
    def get_item(self, item_id: int):
        print(f"DB Client: Fetching item {item_id}")
        if item_id == 42:
            return {"id": item_id, "name": "Widget"}
        return None


def get_db_client():
    return DatabaseClient()


# --- FastAPI Security Dependency ---
def get_current_user(token: str | None = None):
    if token == "secret":
        return {"username": "alice"}
    raise HTTPException(status_code=401, detail="Invalid token")


# --- Route Combining Both ---
@app.get("/items/{item_id}")
async def get_item(
    # FastAPI path parameter
    item_id: Annotated[int, Path(title="The ID of the item to get")],
    # FastAPI security dependency
    current_user: Annotated[dict, Depends(get_current_user)],
    # Picodi dependency using ``wrap=True``
    db: DatabaseClient = Provide(get_db_client, wrap=True),
):
    print(f"User {current_user['username']} requesting item {item_id}")
    item = db.get_item(item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item
```

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

### Resolve Nested Dependencies with Picodi's Inject and Provide

Source: https://github.com/yakimka/picodi/blob/main/docs/topics/dependencies.rst

Illustrates how dependency provider functions can depend on other dependencies using `@inject` and `Provide`. Picodi automatically resolves the entire dependency graph, ensuring providers are called in the correct order.

```python
from picodi import Provide, inject


def get_base_url() -> str:
    return "https://config-service.com"


@inject  # get_api_config depends on get_base_url
def get_api_config(url: str = Provide(get_base_url)) -> dict:
    print(f"Fetching config from {url}")
    # Simulate fetching config based on the URL
    return {"key": "config-key", "timeout": 5}


# Another function can now depend on get_api_config
@inject
def use_config(config: dict = Provide(get_api_config)):
    api_key = config["key"]
    print(f"Using API key: {api_key}")
    return api_key


use_config()
```