### Install Linkd via pip

Source: https://github.com/tandemdude/linkd/blob/master/docs/source/quickstart.md

Use this command to install the library in your Python environment.

```bash
pip install linkd
```

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

### Linkd Dependency Injection Manager Example

Source: https://context7.com/tandemdude/linkd/llms.txt

Demonstrates the basic usage of DependencyInjectionManager to register values and factories, and inject them into an async function within a ROOT context.

```python
import asyncio
import linkd

# Create a manager instance
manager = linkd.DependencyInjectionManager()

# Register dependencies to the ROOT context registry
manager.registry_for(linkd.Contexts.ROOT).register_value(str, "world")
manager.registry_for(linkd.Contexts.ROOT).register_factory(int, lambda: 42)

# Access the default container (available after entering a context)
# manager.default_container  # None until first context is entered

@linkd.inject
async def greet(name: str, number: int) -> str:
    return f"Hello {name}, your number is {number}"

@manager.contextual(linkd.Contexts.ROOT)
async def main() -> None:
    result = await greet()
    print(result)  # Output: Hello world, your number is 42

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

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

### Changelog Fragment Example

Source: https://github.com/tandemdude/linkd/blob/master/fragments/README.md

A sample markdown fragment file for a new feature, named according to the required convention.

```md
Add `foo` method to `Bar` class.
```

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

### Register Factory with Dependencies

Source: https://context7.com/tandemdude/linkd/llms.txt

Illustrates using `register_factory` to create dependencies. The factory function can accept other dependencies, which Linkd will automatically inject when creating the dependency.

```python
import asyncio
import linkd

class Config:
    def __init__(self):
        self.api_key = "secret-key-123"

class ApiClient:
    def __init__(self, config: Config):
        self.api_key = config.api_key

    def call(self, endpoint: str) -> str:
        return f"Calling {endpoint} with key {self.api_key}"

manager = linkd.DependencyInjectionManager()
registry = manager.registry_for(linkd.Contexts.ROOT)

# Register Config as a simple factory
registry.register_factory(Config, Config)

# Register ApiClient with a factory that depends on Config
# The 'config' parameter is automatically injected
registry.register_factory(ApiClient, lambda config: ApiClient(config))

@linkd.inject
async def make_api_call(client: ApiClient) -> str:
    return client.call("/users")

@manager.contextual(linkd.Contexts.ROOT)
async def main() -> None:
    result = await make_api_call()
    print(result)  # Output: Calling /users with key secret-key-123

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

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

### Implement conditional injection with If and Try

Source: https://context7.com/tandemdude/linkd/llms.txt

Use If[T] to handle missing registrations and Try[T] to handle both missing registrations and instantiation failures.

```python
import asyncio
import linkd
from linkd import If, Try

class OptionalCache:
    def get(self, key: str) -> str:
        return f"cached:{key}"

class FlakyService:
    def __init__(self):
        raise RuntimeError("Service unavailable")

manager = linkd.DependencyInjectionManager()
# Intentionally NOT registering OptionalCache
# Register a factory that always fails
manager.registry_for(linkd.Contexts.ROOT).register_factory(FlakyService, FlakyService)

@linkd.inject
async def fetch_data(
    # If[Cache] | None: Returns None if Cache is not registered
    cache: If[OptionalCache] | None,
    # Try[Service] | None: Returns None if creation fails OR not registered
    service: Try[FlakyService] | None,
) -> str:
    cache_status = "available" if cache else "unavailable"
    service_status = "available" if service else "unavailable"
    return f"Cache: {cache_status}, Service: {service_status}"

@manager.contextual(linkd.Contexts.ROOT)
async def main() -> None:
    result = await fetch_data()
    print(result)  # Output: Cache: unavailable, Service: unavailable

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

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

### Registering Prototype Lifetimes

Source: https://context7.com/tandemdude/linkd/llms.txt

Demonstrates registering a factory with a PROTOTYPE lifetime, ensuring a new instance is created for every injection.

```python
registry.register_factory(
    PrototypeCounter,
    PrototypeCounter,
    lifetime=linkd.Lifetime.PROTOTYPE
)

@linkd.inject
async def get_counters(
    singleton1: Counter,
    singleton2: Counter,
    proto1: PrototypeCounter,
    proto2: PrototypeCounter
) -> None:
    print(f"Singleton IDs: {singleton1.id}, {singleton2.id}")  # Same ID
    print(f"Prototype IDs: {proto1.id}, {proto2.id}")  # Different IDs

@manager.contextual(linkd.Contexts.ROOT)
async def main() -> None:
    await get_counters()
    # Output:
    # Singleton IDs: 1, 1
    # Prototype IDs: 2, 3

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

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

### Retrieve dependencies manually from a container

Source: https://context7.com/tandemdude/linkd/llms.txt

Use container.get to manually resolve dependencies and check for their existence within the container.

```python
import asyncio
import linkd

class Database:
    async def query(self) -> str:
        return "query result"

manager = linkd.DependencyInjectionManager()
manager.registry_for(linkd.Contexts.ROOT).register_factory(Database, Database)

async def main() -> None:
    async with manager.enter_context(linkd.Contexts.ROOT) as container:
        # Manually get a dependency from the container
        db = await container.get(Database)
        result = await db.query()
        print(result)  # Output: query result

        # Check if a dependency exists
        print(Database in container)  # Output: True

    await manager.close()

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

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

### Handling Linkd Dependency Injection Exceptions

Source: https://context7.com/tandemdude/linkd/llms.txt

Demonstrates how to catch and handle various exceptions like DependencyNotSatisfiableException, CircularDependencyException, ContainerClosedException, and RegistryFrozenException.

```python
import asyncio
import linkd
from linkd import (
    DependencyInjectionException,
    DependencyNotSatisfiableException,
    CircularDependencyException,
    ContainerClosedException,
    RegistryFrozenException,
)

manager = linkd.DependencyInjectionManager()

# DependencyNotSatisfiableException: Dependency not registered
@linkd.inject
async def needs_unregistered(missing: str) -> None:
    pass

# CircularDependencyException: Dependency requires itself
class ServiceA:
    pass

class ServiceB:
    pass

# This would raise CircularDependencyException if ServiceA required ServiceB
# and ServiceB required ServiceA

# ContainerClosedException: Using a closed container
async def closed_container_example() -> None:
    async with manager.enter_context(linkd.Contexts.ROOT) as container:
        pass  # Container closes here

    # This would raise ContainerClosedException
    # await container.get(str)

# RegistryFrozenException: Modifying registry while containers exist
async def frozen_registry_example() -> None:
    registry = manager.registry_for(linkd.Contexts.ROOT)
    registry.register_value(int, 42)

    async with manager.enter_context(linkd.Contexts.ROOT):
        # This raises RegistryFrozenException - can't modify while container exists
        # registry.register_value(float, 3.14)
        pass

@manager.contextual(linkd.Contexts.ROOT)
async def main() -> None:
    try:
        await needs_unregistered()
    except DependencyNotSatisfiableException as e:
        print(f"Caught: {type(e).__name__}")
        # Output: Caught: DependencyNotSatisfiableException

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

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

### Create Custom Injection Contexts

Source: https://context7.com/tandemdude/linkd/llms.txt

Define and register custom containers to manage dependencies within specific scopes like commands or sessions.

```python
import asyncio
import linkd

# Create a custom context for command handling
class CommandContainer(linkd.Container):
    """Injectable container type for command context."""
    pass

# Register custom context with the global registry
COMMAND_CONTEXT = linkd.global_context_registry.register(
    "myapp.contexts.command",
    CommandContainer  # This type becomes injectable
)

class CommandData:
    def __init__(self, name: str):
        self.name = name

manager = linkd.DependencyInjectionManager()
manager.registry_for(linkd.Contexts.ROOT).register_value(str, "app-name")

@linkd.inject
async def execute_command(
    app_name: str,
    cmd: CommandData,
    container: CommandContainer  # Inject the command container itself
) -> str:
    return f"Executing {cmd.name} in {app_name}"

async def main() -> None:
    async with manager.enter_context(linkd.Contexts.ROOT):
        async with manager.enter_context(COMMAND_CONTEXT) as cmd_container:
            cmd_container.add_value(CommandData, CommandData("deploy"))
            result = await execute_command()
            print(result)  # Output: Executing deploy in app-name

    await manager.close()

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

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

### Singleton vs Prototype Lifetime

Source: https://context7.com/tandemdude/linkd/llms.txt

Explains and demonstrates dependency lifetimes. Singleton ensures a single instance is reused, while Prototype creates a new instance on each request. Singleton is the default behavior.

```python
import asyncio
import linkd

class Counter:
    _count = 0

    def __init__(self):
        Counter._count += 1
        self.id = Counter._count

manager = linkd.DependencyInjectionManager()
registry = manager.registry_for(linkd.Contexts.ROOT)

# SINGLETON: Same instance returned every time (default)
registry.register_factory(
    Counter,
    Counter,
    lifetime=linkd.Lifetime.SINGLETON
)

# Create a separate type for prototype demo
class PrototypeCounter(Counter):
    pass

```

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

### Implement Standalone Dependency Injection

Source: https://github.com/tandemdude/linkd/blob/master/docs/source/quickstart.md

Register dependencies with a manager and use the @inject and @manager.contextual decorators to handle injection.

```python
import asyncio

import linkd

# create a manager instance
manager = linkd.DependencyInjectionManager()
# register a dependency to on of the manager's registries
manager.registry_for(linkd.Contexts.ROOT).register_value(str, "thomm.o")

# enable injection on a function with the inject decorator
@linkd.inject
async def greet(who: str) -> str:
    # the 'who' parameter will be injected by linkd
    return f"hello {who}"


# use the contextual decorator to automatically set up an injection context
@manager.contextual(linkd.Contexts.ROOT)
async def main() -> None:
    # call the injected method
    print(await greet())


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

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

### Integrate Linkd with Quart

Source: https://context7.com/tandemdude/linkd/llms.txt

Utilize ASGI middleware for Quart integration, allowing the standard Linkd injection decorator to function on routes.

```python
import quart
import linkd

class MessageService:
    def get_message(self) -> str:
        return "Hello from Quart!"

manager = linkd.DependencyInjectionManager()
manager.registry_for(linkd.ext.quart.Contexts.ROOT).register_factory(
    MessageService,
    MessageService
)

app = quart.Quart(__name__)
app.asgi_app = linkd.ext.quart.DiContextMiddleware(app.asgi_app, manager)

@app.after_serving
async def shutdown() -> None:
    await manager.close()

@app.route("/")
@linkd.inject  # Standard inject works for Quart!
async def index(service: MessageService) -> str:
    return service.get_message()
```

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

### Register Value with Teardown Function

Source: https://context7.com/tandemdude/linkd/llms.txt

Shows how to register a pre-existing value as a dependency using `register_value`. Includes an optional teardown function that is called when the container is closed, useful for releasing resources like database connections.

```python
import asyncio
import linkd

class DatabaseConnection:
    def __init__(self, url: str):
        self.url = url

    async def close(self):
        print(f"Closing connection to {self.url}")

manager = linkd.DependencyInjectionManager()
registry = manager.registry_for(linkd.Contexts.ROOT)

# Register a value with an optional teardown function
db = DatabaseConnection("postgresql://localhost/mydb")
registry.register_value(
    DatabaseConnection,
    db,
    teardown=lambda conn: conn.close()  # Called when container closes
)

@linkd.inject
async def query_database(db: DatabaseConnection) -> str:
    return f"Connected to {db.url}"

@manager.contextual(linkd.Contexts.ROOT)
async def main() -> None:
    result = await query_database()
    print(result)  # Output: Connected to postgresql://localhost/mydb

async def run():
    await main()
    await manager.close()  # Triggers teardown: "Closing connection to ..."

if __name__ == "__main__":
    asyncio.run(run())
```

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

### Group dependencies using Compose

Source: https://context7.com/tandemdude/linkd/llms.txt

Inherit from linkd.Compose to bundle multiple dependencies into a single class, simplifying function signatures.

```python
import asyncio
import linkd

class Logger:
    def log(self, msg: str) -> None:
        print(f"[LOG] {msg}")

class Database:
    def query(self, sql: str) -> str:
        return f"Result of: {sql}"

class Cache:
    def get(self, key: str) -> str | None:
        return None

# Compose multiple dependencies into a single class
class ServiceDeps(linkd.Compose):
    logger: Logger
    database: Database
    cache: Cache

manager = linkd.DependencyInjectionManager()
registry = manager.registry_for(linkd.Contexts.ROOT)
registry.register_factory(Logger, Logger)
registry.register_factory(Database, Database)
registry.register_factory(Cache, Cache)

# Instead of listing all dependencies individually:
# async def handler(logger: Logger, database: Database, cache: Cache)
# Use a composed dependency:
@linkd.inject
async def handler(deps: ServiceDeps) -> str:
    deps.logger.log("Handling request")
    result = deps.database.query("SELECT * FROM users")
    cached = deps.cache.get("key")
    return result

@manager.contextual(linkd.Contexts.ROOT)
async def main() -> None:
    result = await handler()
    print(result)  # Output: Result of: SELECT * FROM users

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

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

### Integrate Linkd with FastAPI

Source: https://context7.com/tandemdude/linkd/llms.txt

Use the framework-specific decorator and middleware to enable dependency injection in FastAPI applications. Injected parameters must be defined as keyword-only arguments.

```python
import contextlib
import typing as t
import fastapi
import linkd

class UserService:
    def get_user(self, user_id: str) -> dict:
        return {"id": user_id, "name": "John Doe"}

manager = linkd.DependencyInjectionManager()
manager.registry_for(linkd.ext.fastapi.Contexts.ROOT).register_factory(
    UserService,
    UserService
)

@contextlib.asynccontextmanager
async def lifespan(_: fastapi.FastAPI) -> t.AsyncGenerator[None, t.Any]:
    yield
    await manager.close()

app = fastapi.FastAPI(lifespan=lifespan)

# Add middleware to set up DI context for each request
linkd.ext.fastapi.use_di_context_middleware(app, manager)

@app.get("/users/{user_id}")
@linkd.ext.fastapi.inject  # Use framework-specific decorator
async def get_user(
    user_id: str,  # FastAPI path parameter
    *,  # IMPORTANT: linkd-injected params must be keyword-only
    service: UserService,  # Injected by linkd
) -> dict:
    return service.get_user(user_id)
```

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

### Using @manager.contextual Decorator

Source: https://context7.com/tandemdude/linkd/llms.txt

Automatically enters specified DI contexts before executing the decorated async function.

```python
import asyncio
import linkd

class Service:
    def process(self) -> str:
        return "processed"

manager = linkd.DependencyInjectionManager()
manager.registry_for(linkd.Contexts.ROOT).register_factory(Service, Service)

# Automatically enters ROOT context before execution
@manager.contextual(linkd.Contexts.ROOT)
@linkd.inject
async def my_handler(service: Service) -> str:
    return service.process()

# For multiple contexts, list them in order
# @manager.contextual(linkd.Contexts.ROOT, CUSTOM_CONTEXT)

async def main() -> None:
    result = await my_handler()
    print(result)  # Output: processed
    await manager.close()

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

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

### Integrate Linkd with Starlette

Source: https://context7.com/tandemdude/linkd/llms.txt

Configure Starlette applications using the provided middleware and injection decorator for route signature detection.

```python
import contextlib
import typing as t
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route
import linkd

class DataService:
    def fetch(self) -> dict:
        return {"data": "example"}

manager = linkd.DependencyInjectionManager()
manager.registry_for(linkd.ext.starlette.Contexts.ROOT).register_factory(
    DataService,
    DataService
)

@contextlib.asynccontextmanager
async def lifespan(_: Starlette) -> t.AsyncGenerator[None, t.Any]:
    yield
    await manager.close()

@linkd.ext.starlette.inject
async def homepage(request: Request, service: DataService) -> JSONResponse:
    return JSONResponse(service.fetch())

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

middleware = [
    Middleware(linkd.ext.starlette.DiContextMiddleware, manager=manager),
]

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

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

### Managing Context with DependencyInjectionManager

Source: https://context7.com/tandemdude/linkd/llms.txt

Uses a context manager to handle dependency injection scopes, allowing for dynamic addition of dependencies within a specific context.

```python
import asyncio
import linkd

class RequestData:
    def __init__(self, request_id: str):
        self.request_id = request_id

manager = linkd.DependencyInjectionManager()
manager.registry_for(linkd.Contexts.ROOT).register_value(str, "app-config")

@linkd.inject
async def handle_request(config: str, req: RequestData) -> str:
    return f"Request {req.request_id} with config: {config}"

async def main() -> None:
    # Enter ROOT context first
    async with manager.enter_context(linkd.Contexts.ROOT) as root_container:
        # Add request-specific dependency to the container
        root_container.add_value(RequestData, RequestData("req-001"))

        result = await handle_request()
        print(result)  # Output: Request req-001 with config: app-config

    # Context is now closed, dependencies cleaned up
    await manager.close()

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

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

### Add ephemeral dependencies to a container

Source: https://context7.com/tandemdude/linkd/llms.txt

Use add_value and add_factory to register dependencies that exist only within the scope of a specific container.

```python
import asyncio
import linkd

class GlobalConfig:
    env = "production"

class RequestContext:
    def __init__(self, user_id: str):
        self.user_id = user_id

manager = linkd.DependencyInjectionManager()
manager.registry_for(linkd.Contexts.ROOT).register_factory(GlobalConfig, GlobalConfig)

@linkd.inject
async def process(config: GlobalConfig, ctx: RequestContext) -> str:
    return f"User {ctx.user_id} in {config.env}"

async def main() -> None:
    async with manager.enter_context(linkd.Contexts.ROOT) as container:
        # Add request-specific value to this container only
        container.add_value(RequestContext, RequestContext("user-123"))

        # Add a factory for request-scoped dependencies
        container.add_factory(
            str,
            lambda ctx: f"Request from {ctx.user_id}",
        )

        result = await process()
        print(result)  # Output: User user-123 in production

    await manager.close()

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

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

### Using @linkd.inject Decorator

Source: https://context7.com/tandemdude/linkd/llms.txt

Enables dependency injection on functions. Use linkd.INJECTED as a default value for parameters to satisfy type checkers.

```python
import asyncio
import linkd

class Logger:
    def log(self, msg: str) -> None:
        print(f"[LOG] {msg}")

class UserService:
    def __init__(self):
        self.users = {"1": "Alice", "2": "Bob"}

    def get_user(self, user_id: str) -> str:
        return self.users.get(user_id, "Unknown")

manager = linkd.DependencyInjectionManager()
registry = manager.registry_for(linkd.Contexts.ROOT)
registry.register_factory(Logger, Logger)
registry.register_factory(UserService, UserService)

# Use linkd.INJECTED to satisfy type checkers when calling without args
@linkd.inject
async def process_user(
    user_id: str,  # Regular parameter - must be provided
    logger: Logger = linkd.INJECTED,  # Injected parameter
    service: UserService = linkd.INJECTED  # Injected parameter
) -> str:
    user = service.get_user(user_id)
    logger.log(f"Processing user: {user}")
    return user

@manager.contextual(linkd.Contexts.ROOT)
async def main() -> None:
    # Only provide user_id; logger and service are injected
    result = await process_user("1")
    print(f"Result: {result}")  # Output: Result: Alice

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

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