Creating Pico-Boot Plugins¶

# src/my_plugin/__init__.py
"""
My Pico-Boot Plugin

Provides integration with SomeService.
"""

# Re-export main components for easy access
from .config import MyPluginConfig
from .components import MyPluginService
from .providers import build_some_client

__all__ = [
    "MyPluginConfig",
    "MyPluginService",
    "build_some_client",
]

# src/my_plugin/config.py
from dataclasses import dataclass
from pico_ioc import configured

@configured(prefix="my_plugin")
@dataclass
class MyPluginConfig:
    """
    Configuration for my-plugin.

    Add to your application.yaml:

        my_plugin:
          api_key: your-api-key
          timeout: 30
          enabled: true

    Or use environment variables:
        MY_PLUGIN_API_KEY=your-api-key
        MY_PLUGIN_TIMEOUT=30
        MY_PLUGIN_ENABLED=true
    """
    api_key: str
    timeout: int = 30
    enabled: bool = True

# src/my_plugin/components.py
from pico_ioc import component, health, cleanup
from .config import MyPluginConfig

@component
class MyPluginService:
    """Main service provided by the plugin."""

    def __init__(self, config: MyPluginConfig):
        self.config = config
        self._connected = False

    def connect(self) -> None:
        if self.config.enabled:
            # Connection logic
            self._connected = True

    def do_something(self) -> str:
        if not self.config.enabled:
            return "Plugin disabled"
        return "Plugin working!"

    @health
    def is_healthy(self) -> bool:
        """Health check for observability."""
        return self._connected or not self.config.enabled

    @cleanup
    def close(self) -> None:
        """Cleanup when container shuts down."""
        self._connected = False

# src/my_plugin/providers.py
from pico_ioc import provides
from .config import MyPluginConfig

# Example: providing a third-party client
from some_external_library import SomeClient

@provides(SomeClient)
def build_some_client(config: MyPluginConfig) -> SomeClient:
    """
    Provides a configured SomeClient instance.

    This allows applications to inject SomeClient directly
    without knowing how to configure it.
    """
    return SomeClient(
        api_key=config.api_key,
        timeout=config.timeout
    )

# src/my_plugin/__init__.py
from pico_ioc import CustomScanner
from types import ModuleType

class MyCustomScanner(CustomScanner):
    """Discovers components with @my_decorator."""

    def scan(self, module: ModuleType) -> None:
        for name in dir(module):
            obj = getattr(module, name)
            if hasattr(obj, "_my_plugin_marker"):
                # Register with container
                self.register_component(obj)

# Export for Pico-Boot to discover
PICO_SCANNERS = [MyCustomScanner()]

"""
Pico-Redis: Redis integration for Pico-Boot

Usage:
    1. Install: pip install pico-redis
    2. Add to application.yaml:
        redis:
          url: redis://localhost:6379/0
    3. Inject redis.Redis in your components:
        @component
        class MyService:
            def __init__(self, redis: redis.Redis):
                self.redis = redis
"""

from .config import RedisConfig
from .providers import build_redis_client

__all__ = ["RedisConfig", "build_redis_client"]

from dataclasses import dataclass
from pico_ioc import configured

@configured(prefix="redis")
@dataclass
class RedisConfig:
    """Redis connection configuration."""
    url: str = "redis://localhost:6379/0"
    max_connections: int = 10
    decode_responses: bool = True
    socket_timeout: float = 5.0

import redis
from pico_ioc import provides, cleanup
from .config import RedisConfig

_pool: redis.ConnectionPool | None = None

@provides(redis.Redis)
def build_redis_client(config: RedisConfig) -> redis.Redis:
    """Provides a configured Redis client with connection pooling."""
    global _pool

    if _pool is None:
        _pool = redis.ConnectionPool.from_url(
            config.url,
            max_connections=config.max_connections,
            decode_responses=config.decode_responses,
            socket_timeout=config.socket_timeout,
        )

    return redis.Redis(connection_pool=_pool)

@cleanup
def close_redis_pool() -> None:
    """Cleanup connection pool on shutdown."""
    global _pool
    if _pool is not None:
        _pool.disconnect()
        _pool = None

# tests/test_plugin.py
import pytest
from pico_ioc import init, configuration, DictSource

def test_plugin_loads():
    """Verify plugin components are discoverable."""
    from my_plugin import MyPluginConfig, MyPluginService

    config = configuration(
        DictSource({"my_plugin": {"api_key": "test-key"}})
    )

    container = init(
        modules=["my_plugin"],
        config=config
    )

    service = container.get(MyPluginService)
    assert service.config.api_key == "test-key"
    container.shutdown()

# tests/test_integration.py
import os
import pytest

def test_auto_discovery():
    """Verify plugin is discovered by pico-boot."""
    os.environ["MY_PLUGIN_API_KEY"] = "test-key"

    from pico_boot import init
    from my_plugin import MyPluginService

    container = init(modules=[])  # Empty - relies on auto-discovery

    # Plugin should be loaded automatically
    service = container.get(MyPluginService)
    assert service is not None

    container.shutdown()
    del os.environ["MY_PLUGIN_API_KEY"]

# Good - unique prefix
@configured(prefix="my_plugin")

# Bad - too generic, may conflict
@configured(prefix="database")

@dataclass
class MyPluginConfig:
    required_field: str          # No default = required
    optional_field: int = 30     # With default = optional

@configured(prefix="my_plugin")
@dataclass
class MyPluginConfig:
    """
    My Plugin Configuration.

    YAML:
        my_plugin:
          api_key: your-key
          timeout: 30

    Environment:
        MY_PLUGIN_API_KEY=your-key
        MY_PLUGIN_TIMEOUT=30
    """

@component
class MyService:
    @health
    def is_healthy(self) -> bool:
        return self._connection.is_alive()

@component
class MyService:
    @cleanup
    async def close(self) -> None:
        await self._connection.close()

try:
    import optional_lib
    HAS_OPTIONAL = True
except ImportError:
    HAS_OPTIONAL = False

if HAS_OPTIONAL:
    @provides(optional_lib.Client)
    def build_client() -> optional_lib.Client:
        return optional_lib.Client()

@component(profiles=["production"])
class ProductionCache:
    pass

@component(profiles=["development", "test"])
class MockCache:
    pass

@configured(prefix="my_plugin")  # <-- Same prefix