Skip to main content

No project description provided

Project description

init-provider

Initialization and instance provider framework for Python.

PyPI - Version PyPI - Python Version PyPI - Status PyPI - License

Use cases

  • Solve initialization hell: Declare what depends on what and forget about it!
  • Share object instances: Expose a reusable instance of Settings or a Connection Pool.
  • Business logic: Implement clean internal APIs.
  • Entry point: Define an entry point for your CLI, Web API, background worker, etc.

Quick Start

Runnable end‑to‑end example:

from init_provider import BaseProvider, requires

class Config(BaseProvider):
    message: str

    def provider_init(self) -> None:
        self.message = "Hello"

@requires(Config)
class Greeter(BaseProvider):
    def greet(self, name: str) -> str:
        return f"{Config.message}, {name}!"

if __name__ == "__main__":
    print(Greeter.greet("World"))

Installation

  • Available on PyPI.
  • Pure Python with zero runtime dependencies.
  • Supports Python 3.10+.
# Using pip
pip install init-provider

# Using uv
uv add init-provider

Design patterns

Write clean, testable, and maintainable code. init-provider lets you implement any of the common design patterns below in a very concise way:

  • Repository: Abstract the data access layer (S3, SQL, REST, etc) and return Models.
  • Controller: Modify internal state based on requests from user or other systems.
  • Service: Implement business logic.
  • Singleton: Provide a single instance of a class, such as Settings.

Usage

Providers are just classes. In fact, they look a lot like a dataclass but with three major differences:

  1. You do not need to instantiate the provider class.
  2. Providers can depend on each other.
  3. Calling any method or attribute of a provider will trigger initialization.

Inherit BaseProvider

Create a class that inherits from BaseProvider. This automatically registers your provider inside the framework.

from init_provider import BaseProvider

class WeatherProvider(BaseProvider):
    """Fetch weather data from the API."""

Store state in class variables

Use class variables just like you would in a dataclass.

# ...
class WeatherProvider(BaseProvider):
    # ...
    _base_url: str = "https://theweather.com/api"

Note: init_provider doesn't care about underscores in variable and method names. It will expose them all the same.

Initialize inside provider_init()

When you need to initialize the provider, you can focus on what needs to be initialized rather than when it needs to be initialized.

Not all providers require initialization, but when they do, you can define it inside the provider_init() method.

For example, you might want to initialize a reusable aiohttp session during runtime, when the asyncio event loop is already running.

# ...
import asyncio
from aiohttp import ClientSession

class WeatherProvider(BaseProvider):
    # ...
    _session: ClientSession

    def provider_init(self) -> None:
        self._session = ClientSession()

if __name__ == "__main__":
    if WeatherProvider._session.closed:
        print("Session is still closed")

Note 1: in the example aboev, the _session variable is declared without a value. The initialization is done inside the provider_init(). Trying to access the _session object will trigger the initialization chain.

Note 2: The provider_init method of the owner class is the only place where initialization will not be triggered, when the object is accessed.

Warning: Declaring a class variable with a default value will mean that it's

Add business logic

Providers are great for encapsulating reusable business logic in a methods. Every method of the provider automatically becomes a guarded method. Guarded methods cause initialization of the provider chain, when they are called.

Note: Reserved methods that contain double underscore (__) and methods decorated with @staticmethod or @classmethod will not be guarded.

from init_provider import BaseProvider

class WeatherProvider(BaseProvider):
    # ...
    @classmethod
    def get_url(cls, path: str) -> str:
        return f"{cls._base_url}/{path}"

Specify dependencies with @requires

Use the @requires decorator to list other providers that the WeatherProvider depends on.

@requires(GeoProvider)
class WeatherProvider(BaseProvider):
    # ...

Finally, guard the class methods that need everything to be initialized before they are called with the @guarded decorator.

@requires(GeoProvider)
class WeatherProvider(BaseProvider):
    # ...
    @guarded
    def get_weather(cls, city: str) -> dict:
        return cls._session.get(cls.get_url(f"weather?q={city}")).json()

Examples

Weather service

import asyncio
import logging
from aiohttp import ClientSession
from init_provider import BaseProvider, requires

logging.basicConfig(level=logging.DEBUG, format="%(levelname)-8s %(message)s")

class GeoService(BaseProvider):
    def city_coordinates(self, name: str) -> tuple[float, float]:
        """Returns the latitude and longitude of a city."""
        if name == "London":
            return 51.509, -0.118  # London, UK
        elif name == "New York":
            return 40.7128, -74.0060  # New York, USA
        raise ValueError(f"Unknown city: {name}")

@requires(GeoService)
class WeatherService(BaseProvider):
    _session: ClientSession
    _base_url: str = "https://api.open-meteo.com/v1/forecast/"
    
    def provider_init(self) -> None:
        # Properly initializing aiohttp session at runtime, when the
        # default asyncio loop is already running.
        self._session = ClientSession(self._base_url)

    @classmethod
    async def close(cls):
        await cls._session.close()

    async def temperature(self, city: str) -> float:
        lat, lon = GeoService.city_coordinates(city)
        params = {"latitude": lat, "longitude": lon, "hourly": "temperature_2m"}
        async with self._session.get(self._base_url, params=params) as resp:
            data = await resp.json()
            return data["hourly"]["temperature_2m"][0]

async def main():
    # This will immediately initialize WeatherService and its dependencies,
    # because we have attempted to access the _session property.
    print(f"Is session closed: {WeatherService._session.closed}")

    # Subsequent calls do not reinitialize the provider.
    london = await WeatherService.temperature('London')
    new_york = await WeatherService.temperature('New York')
    print(f"London: {london:.2f}°C")
    print(f"New York: {new_york:.2f}°C")

    # Release the resources. Normally, this would be implemented in the
    # provider_dispose() method of the provider, but the async client must be closed
    # inside the same event loop it was created.
    await WeatherService.close()
    print(f"Is session closed: {WeatherService._session.closed}")


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

Output:

$ uv run python examples/weather_service.py
DEBUG    Using selector: KqueueSelector
DEBUG    About to initialize provider WeatherService because of: _session
DEBUG    Initialization order for provider WeatherService is: GeoService, WeatherService
DEBUG    Initializing provider GeoService...
INFO     Provider GeoService initialized
DEBUG    Initializing provider WeatherService...
INFO     Provider WeatherService initialized
Is session closed: False
London: 13.10°C
New York: 11.30°C
Is session closed: True
DEBUG    Provider dispose call order: ['WeatherService', 'GeoService']
INFO     Dispose hook for WeatherService was executed.
INFO     Dispose hook for GeoService was executed.

User service

import logging
import os
import sqlite3
import warnings
from contextlib import contextmanager
from typing import Generator

from init_provider import BaseProvider, requires, setup

# (Optional) Declare a setup function to be executed once per application
# process before any provider is initialized.
@setup
def configure():
    log_format = "%(levelname)-8s %(message)s"
    logging.basicConfig(level=logging.DEBUG, format=log_format)
    warnings.filterwarnings("ignore", module="some_module")

# ↓ Basic provider. Exposes 1 attribute: connection 
class DatabaseService(BaseProvider):
    """Single instance of connection ot SQLite."""

    # ↓ Any attempt to access a provider attribute outside
    #   of provider_init() will cause the provider to be initialized.
    db_path: str

    # ↓ Initialize, just like in a dataclass. But you NEVER
    #   have to create an instance of a provider manually.
    def provider_init(self) -> None:
        # Run some one-time initialization logic
        self.db_path = "database.db"

        # Initialize the database. This will only be done once
        # across the entire lifecycle of the application.
        with sqlite3.connect(self.db_path) as conn:
            cur = conn.cursor()
            # Create a table
            cur.execute(
                "CREATE TABLE IF NOT EXISTS users "
                "(id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT)"
            )
            # Add mock data
            cur.executemany(
                "INSERT INTO users (name) VALUES (?)",
                [("Alice",), ("Bob",)],
            )
            conn.commit()


    # ↓ Declare a dispose method to be called before the application exits.
    def provider_dispose(self):
        os.unlink(self.db_path)

    # ↓ Any call to the `conn` method will cause the
    #   provider to be initialized, if not already done.
    @contextmanager
    def conn(self) -> Generator[sqlite3.Connection, None, None]:
        """One-time connection to the database."""
        with sqlite3.connect(self.db_path) as conn:
            yield conn

# ↓ This one depends on another provider.
@requires(DatabaseService)
class UserService(BaseProvider):
    """Intenal API class to abstract the Users data layer."""

    # → Notice: NO provider_init() method here! Because there is nothing
    #   to initialize inside this specific provider itself.

    # ↓ Require initialization of all dependencies when this
    #   method is called.
    def get_name(self, user_id: int) -> str | None:
        """Get user name based on ID"""

        # ↓ Access the method from another provider
        with DatabaseService.conn() as conn:
            cur = conn.cursor()
            if result := cur.execute(
                "SELECT name FROM users WHERE id = ?", (user_id,)
            ).fetchone():
                return result[0]
            else:
                return None

if __name__ == "__main__":
    # ↓ This will cause the chain of dependencies to be
    #   initialized in the following order:
    #   1. configure() function will be called
    #   2. DatabaseService
    database_path = DatabaseService.db_path
    print(f">> {database_path}")

    # ↓ This will only initialize the UserService, because
    #   its dependencies are already initialized.
    user_1 = UserService.get_name(1)
    print(f">> {user_1}")

    # ↓ Let's get the name of another user. NOTHING extra will be
    #   done because the dependency graph is already initialized.
    user_2 = UserService.get_name(2)
    print(f">> {user_2}")

Output:

$ uv run python examples/user_service.py
INFO     Setup hook executed.
DEBUG    About to initialize provider DatabaseService because of: db_path
DEBUG    Initialization order for provider DatabaseService is: DatabaseService
DEBUG    Initializing provider DatabaseService...
INFO     Provider DatabaseService initialized
DEBUG    About to initialize provider UserService because of: get_name
DEBUG    Initialization order for provider UserService is: DatabaseService (initialized), UserService
DEBUG    Initializing provider UserService...
INFO     Provider UserService initialized
>> database.db
>> Alice
>> Bob
DEBUG    Provider dispose call order: ['UserService', 'DatabaseService']
INFO     Dispose hook for UserService was executed.
INFO     Dispose hook for DatabaseService was executed.

Troubleshooting

Enable logging

The framework produces logs tied to the init_provider module. Make sure the logs from this module are not suppressed in the global logging configuration.

The easiest way to enable logging is to set the logging level to DEBUG:

import logging
logging.basicConfig(level=logging.DEBUG)

Which will allow you to see what init_provider is doing:

$ uv run python examples/weather_service.py
DEBUG    About to initialize provider WeatherService because of: session
DEBUG    Initialization order for provider WeatherService is: GeoService, WeatherService
DEBUG    Initializing provider GeoService...
INFO     Provider GeoService initialized successfully
DEBUG    Initializing provider WeatherService...

License

Licensed under the Apache-2.0 License.

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

init_provider-0.1.1.tar.gz (25.3 kB view details)

Uploaded Source

Built Distribution

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

init_provider-0.1.1-py3-none-any.whl (20.5 kB view details)

Uploaded Python 3

File details

Details for the file init_provider-0.1.1.tar.gz.

File metadata

  • Download URL: init_provider-0.1.1.tar.gz
  • Upload date:
  • Size: 25.3 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: uv/0.9.0

File hashes

Hashes for init_provider-0.1.1.tar.gz
Algorithm Hash digest
SHA256 37d8e9e23398991725645f00cb4724ed2b70e269b538cae10d1c162d7d7db186
MD5 f86a4e9c48a3f7ae1682bbd7ebdd7e4a
BLAKE2b-256 432028d049cd49c2faefbeca9cde5f963f7db61f14c575d8797224dd5ba93efa

See more details on using hashes here.

File details

Details for the file init_provider-0.1.1-py3-none-any.whl.

File metadata

File hashes

Hashes for init_provider-0.1.1-py3-none-any.whl
Algorithm Hash digest
SHA256 29640502722b751bf82c50b7f7ed7c0a17793ce3755fd3b10884d2c619d6b97f
MD5 770052172bd5d8e341d8afd889a43707
BLAKE2b-256 9bbf5484fc282bbfaabe7597a82cf7dc631e536e46f187b78e24178f261d19fe

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