Skip to main content

Lightweight async dependency injection framework for Python. Inspired by FastAPI's Depends.

Project description

Egg

Lightweight async dependency injection framework for Python. Inspired by FastAPI's Depends, but framework-agnostic.

Installation

Requires Python >= 3.12

pip install injegg
# or
uv add injegg

Usage

Wrap dependencies in Egg() and use the @hatch_eggs decorator to auto-resolve them.

Using Annotated type hints

import asyncio
from typing import Annotated
from egg import Egg, hatch_eggs

async def get_client():
    return "http_client"

@hatch_eggs
async def main(client: Annotated[str, Egg(get_client)]):
    print(client)  # "http_client"

asyncio.run(main())

Using default parameter syntax

@hatch_eggs
async def main(client: str = Egg(get_client)):
    print(client)  # "http_client"

Injecting into class methods

class UserService:
    @hatch_eggs
    async def get_user(self, db: Annotated[Database, Egg(get_database)]):
        return await db.query("SELECT * FROM users")

service = UserService()
asyncio.run(service.get_user())

Nested dependencies

Dependencies can depend on other dependencies. They resolve in the correct order.

async def get_config():
    return {"db_url": "postgres://localhost/mydb"}

async def get_database(config: Annotated[dict, Egg(get_config)]):
    return Database(config["db_url"])

async def get_user_repo(db: Annotated[Database, Egg(get_database)]):
    return UserRepository(db)

@hatch_eggs
async def main(repo: Annotated[UserRepository, Egg(get_user_repo)]):
    users = await repo.find_all()
    print(users)

asyncio.run(main())

Multiple dependencies

@hatch_eggs
async def handler(
    db: Annotated[Database, Egg(get_database)],
    cache: Annotated[Cache, Egg(get_cache)],
    logger: Annotated[Logger, Egg(get_logger)],
):
    logger.info("Fetching data")
    data = await cache.get("key") or await db.query("...")
    return data

Dependencies

A dependency is any callable. Parameters are resolved by name from context.

Supported types

# Async function
async def get_client():
    return AsyncClient()

# Sync function
def get_config():
    return load_config()

# Class instance with __call__
class TokenProvider:
    def __init__(self, secret: str):
        self.secret = secret

    async def __call__(self, user_id: str):
        return generate_token(user_id, self.secret)

# Usage: pass config at instantiation, user_id comes from context
token_provider = TokenProvider(secret="abc123")

@hatch_eggs
async def handler(
    user_id: str,
    token: Annotated[str, Egg(token_provider)],  # calls token_provider(user_id=...)
):
    ...

How It Works

  1. Decoration: The @hatch_eggs decorator wraps your async function
  2. Inspection: At call time, it inspects type hints (Annotated[T, Egg(...)]) and default values (= Egg(...)) to find dependencies
  3. Resolution: For each Egg, the dependency function is called. If that function also has Egg parameters, they're resolved first (recursive)
  4. Caching: Results are cached by default—calling the same dependency twice returns the cached value
  5. Injection: Resolved values replace the Egg markers and your function is called with the real dependencies
@hatch_eggs
async def main(db: Annotated[Database, Egg(get_db)]):
                                        ↓
                              calls get_db()
                                        ↓
                              caches result
                                        ↓
                              main(db=<Database>)

Auto-injection from context

Once a dependency is resolved, it's added to the available context by parameter name. Nested dependencies can then receive it automatically without an Egg() wrapper—just match the parameter name:

async def get_database():
    return Database("postgres://localhost/mydb")

# No Egg() needed—"db" is auto-injected from context
async def get_user_repo(db):
    return UserRepository(db)

@hatch_eggs
async def main(
    db: Annotated[Database, Egg(get_database)],        # resolved first, added to context as "db"
    repo: Annotated[UserRepository, Egg(get_user_repo)]  # get_user_repo receives "db" automatically
):
    users = await repo.find_all()

This enables implicit wiring—dependencies don't need to know how their own dependencies are created.

Forwarding caller arguments to dependencies

Arguments passed by the caller are also available for injection into dependencies:

async def get_api_client(config_id: str):
    config = await load_config(config_id)
    return AsyncClient(base_url=config["api_url"])

@hatch_eggs
async def create(config_id: str, api_client: AsyncClient = Egg(get_api_client)):
    return await api_client.post("/resources")

# config_id is passed to create(), then auto-injected into get_api_client()
await create(config_id="production")

The flow:

  1. create("production") is called with config_id="production"
  2. config_id is added to the available context
  3. get_api_client needs config_id—it's auto-injected from context
  4. create receives the fully configured api_client

Circular dependencies are detected and raise EggHatchingError.

Use Cases

Testing with mocks

Easily swap dependencies in tests by passing them directly—no patching needed:

@hatch_eggs
async def process_payment(gateway: PaymentGateway = Egg(get_payment_gateway)):
    return await gateway.charge(100)

# In tests: bypass the Egg by passing the dependency directly
async def test_process_payment():
    mock_gateway = MockPaymentGateway()
    result = await process_payment(gateway=mock_gateway)
    assert result.success

Background jobs and scripts

Clean dependency setup for CLI tools, background workers, or one-off scripts:

@hatch_eggs
async def run_sync_job(
    tenant_id: str,
    db: Annotated[Database, Egg(get_database)],
    api: Annotated[ExternalAPI, Egg(get_external_api)],
):
    records = await api.fetch_updates(tenant_id)
    await db.bulk_upsert(records)

# Simple invocation with just the business parameter
asyncio.run(run_sync_job(tenant_id="acme-corp"))

Contributing

  1. Fork the repository
  2. Clone your fork and install dependencies:
    git clone https://github.com/your-username/egg.git
    cd egg
    uv sync
    
  3. Make your changes
  4. Run tests:
    pytest egg/tests.py -v
    
  5. Submit a pull request

License

MIT

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

injegg-0.1.0.tar.gz (10.4 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

injegg-0.1.0-py3-none-any.whl (9.8 kB view details)

Uploaded Python 3

File details

Details for the file injegg-0.1.0.tar.gz.

File metadata

  • Download URL: injegg-0.1.0.tar.gz
  • Upload date:
  • Size: 10.4 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: uv/0.7.21

File hashes

Hashes for injegg-0.1.0.tar.gz
Algorithm Hash digest
SHA256 17221e0f269b514336d6e19eb3e988620e908e51fe02bea39bc61f0591dd2058
MD5 c2bdd8d4f714cd70c6cda4d5c15274d0
BLAKE2b-256 7ade6e4f40a9a358ffd9af4364085b09343f5a4594ceb786ad9cd7ee686f15d7

See more details on using hashes here.

File details

Details for the file injegg-0.1.0-py3-none-any.whl.

File metadata

  • Download URL: injegg-0.1.0-py3-none-any.whl
  • Upload date:
  • Size: 9.8 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: uv/0.7.21

File hashes

Hashes for injegg-0.1.0-py3-none-any.whl
Algorithm Hash digest
SHA256 1c00ab3d5a9aed9fe911f94853e69487ba8951832e9a1256fccbbbdd2c6969a2
MD5 28f8726e996691f7256c3cef23b3b8c7
BLAKE2b-256 f4b2bd1cb1ca1dfd884d292111c3e3dd266dd1fa8191f3cbdcc358688075d2f4

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page