Advanced: The Event Bus¶

@component
class UserService:
    def __init__(
        self,
        db: Database,
        email_service: EmailService,
        analytics: AnalyticsService,
        audit_log: AuditLogService
    ):
        self.db = db
        self.email_service = email_service
        self.analytics = analytics
        self.audit_log = audit_log

    async def create_user(self, email: str):
        # 1. Business Logic
        user = self.db.save(email)

        # 2. Tightly Coupled Calls
        # What if EmailService fails?
        # What if we add a new service? We have to edit this file.
        await self.email_service.send_welcome_email(user)
        await self.analytics.track_event("user_created", user.id)
        await self.audit_log.record(f"User {user.id} created")

        return user

from dataclasses import dataclass
from pico_ioc.event_bus import Event

@dataclass
class UserCreatedEvent(Event):
    user_id: int
    email: str

# events.py
from dataclasses import dataclass
from pico_ioc.event_bus import Event

@dataclass
class UserCreatedEvent(Event):
    user_id: int
    email: str

# services/user_service.py
from pico_ioc import component
from pico_ioc.event_bus import EventBus
from .events import UserCreatedEvent

# Assume Database is defined elsewhere
class Database:
    def save(self, email: str) -> 'User': ...  # Mock

class User:
    id: int
    email: str

@component
class UserService:
    def __init__(self, db: Database, bus: EventBus):
        self.db = db
        self.bus = bus  # Just inject the bus

    async def create_user(self, email: str):
        # 1. Business Logic
        user = self.db.save(email)  # Assume save returns a User object with id and email

        # 2. Publish Event
        # We just shout "this happened!" and don't care who is listening.
        event = UserCreatedEvent(user_id=user.id, email=user.email)
        await self.bus.publish(event)

        return user

# services/email_service.py
import asyncio  # For simulation
from pico_ioc import component
from pico_ioc.event_bus import AutoSubscriberMixin, subscribe
from ..events import UserCreatedEvent

@component
class EmailService(AutoSubscriberMixin):

    @subscribe(UserCreatedEvent)
    async def on_user_created(self, event: UserCreatedEvent):
        # This method is automatically called when UserCreatedEvent is published
        print(f"EMAIL: Sending welcome email to {event.email}")
        await self.send_email(event.email, "Welcome!")

    async def send_email(self, to, body):
        await asyncio.sleep(0.01)  # Simulate sending
        pass

# services/analytics_service.py
import asyncio  # For simulation
from pico_ioc import component
from pico_ioc.event_bus import AutoSubscriberMixin, subscribe, ExecPolicy
from ..events import UserCreatedEvent

@component
class AnalyticsService(AutoSubscriberMixin):

    @subscribe(UserCreatedEvent, policy=ExecPolicy.TASK)
    async def on_user_created(self, event: UserCreatedEvent):
        # This handler is non-critical, so we run it as a
        # "fire and forget" task that doesn't block the publisher.
        print(f"ANALYTICS: Tracking event for user {event.user_id}")
        await self.track(event.user_id)

    async def track(self, user_id):
        await asyncio.sleep(0.02)  # Simulate tracking
        pass

# main.py
import asyncio
import pico_ioc.event_bus  # Import the module to be scanned
from pico_ioc import init, component

# Adjust imports based on your actual project structure
# Example assumes services are in ./app/services/ and events in ./app/events/
from app.services.user_service import UserService
from app.services.email_service import EmailService  # Need these for scanning
from app.services.analytics_service import AnalyticsService
from app.events import UserCreatedEvent

# Mock Database for runnable example
class MockUser:
    id: int = 1
    email: str = "test@example.com"

@component
class MockDatabase:
    def save(self, email: str) -> MockUser:
        print(f"DB: Saving {email}")
        return MockUser()

container = init(
    modules=[
        "app.events",  # Contains UserCreatedEvent definition
        "app.services.user_service",
        "app.services.email_service",
        "app.services.analytics_service",
        pico_ioc.event_bus,  # Don't forget this!
        __name__,  # Include current module for MockDatabase
    ]
)

async def run_example():
    user_service = await container.aget(UserService)
    print("Creating user...")
    await user_service.create_user("alice@example.com")
    print("User creation initiated (event published).")

    # Give background tasks (like analytics) a moment to run in this example
    await asyncio.sleep(0.1)

    print("Cleaning up...")
    await container.cleanup_all_async()

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

# --- Example Output ---
# Creating user...
# DB: Saving alice@example.com
# EMAIL: Sending welcome email to test@example.com
# ANALYTICS: Tracking event for user 1
# User creation initiated (event published).
# Cleaning up...

from pico_ioc.event_bus import EventBus

bus = await container.aget(EventBus)
await bus.start_worker()  # starts the background task to drain the queue

# ... now bus.post(event) will enqueue and be processed in the background ...

await bus.stop_worker()   # gracefully stops the worker (e.g., on shutdown)