<h1><code>init-provider</code></h1>
Initialization and instance provider framework for Python.
* **Init order**: for example, `ProviderA` depends on `ProviderB`.
* **Reusable objects**: expose instances of Settings or a Connection Pool.
* **Business logic**: clean internal APIs.
* **Entry point**: use for a CLI, Web API, background worker, etc.
- [Quick start](#quick-start)
- [Installation](#installation)
- [Usage](#usage)
- [Inherit from `BaseProvider`](#inherit-from-baseprovider)
- [Store in class variables](#store-in-class-variables)
- [Initialize using `__init__()`](#initialize-using-__init__)
- [Dispose using `__del__()`](#dispose-using-__del__)
- [Decorate methods using `@init`](#decorate-methods-using-init)
- [Dependencies using `@requires`](#dependencies-using-requires)
- [Examples](#examples)
- [Weather service](#weather-service)
- [User service](#user-service)
- [Troubleshooting](#troubleshooting)
- [Enable logging](#enable-logging)
- [Design choices](#design-choices)
- [License](#license)
## Quick start
Below is a full runnable example that uses everything in this library.
```python
import logging
from pathlib import Path
from init_provider import BaseProvider, init, requires, setup, dispose
@setup
def configure() -> None:
logging.basicConfig(
level=logging.DEBUG,
format="%(levelname)-8s %(name)-15s %(message)s",
)
if not Path("file.txt").exists():
logging.info("> file.txt does not yet exist")
@dispose
def cleanup() -> None:
if not Path("file.txt").exists():
logging.info("> file.txt no longer exist")
class Storage(BaseProvider):
path = Path("file.txt")
def __init__(self):
logging.info("> create Storage")
self.path.touch()
def __del__(self):
logging.info("> dispose of Storage")
self.path.unlink()
@init
def write(self, content: str) -> None:
logging.info(f"> write to Storage: {content}")
self.path.write_text(content)
@init
def read(self) -> str:
data = self.path.read_text()
logging.info(f"> read from Storage: {data}")
return data
@requires(Storage)
class Namer(BaseProvider):
def __init__(self) -> None:
logging.info("> create Namer")
Storage.write("Bobby")
@requires(Namer)
class Greeter(BaseProvider):
define_at_runtime: str
def __init__(self):
logging.info("> create Greeter")
self.define_at_runtime = Storage.read()
@init
def greet(self) -> None:
print(f">>> Hello, {self.define_at_runtime}!")
if __name__ == "__main__":
Greeter.greet()
```
Output:
```shell
$ uv run python examples/full_example.py
INFO root > file.txt does not yet exist
INFO init_provider Setup hook executed.
DEBUG init_provider About to initialize provider Greeter because of: greet
DEBUG init_provider Initialization order for provider Greeter is: Storage, Namer, Greeter
DEBUG init_provider Initializing provider Storage...
INFO root > create Storage
INFO init_provider Provider Storage initialized
DEBUG init_provider Initializing provider Namer...
INFO root > create Namer
INFO root > write to Storage: Bobby
INFO init_provider Provider Namer initialized
DEBUG init_provider Initializing provider Greeter...
INFO root > create Greeter
INFO root > read from Storage: Bobby
INFO init_provider Provider Greeter initialized
>>> Hello, Bobby!
DEBUG init_provider Provider dispose call order: ['Greeter', 'Namer', 'Storage']
INFO init_provider Dispose hook for Greeter was executed.
INFO init_provider Dispose hook for Namer was executed.
INFO root > dispose of Storage
INFO init_provider Dispose hook for Storage was executed.
INFO root > file.txt no longer exist
INFO init_provider Dispose hook executed.
```
## Installation
Using `pip`:
```shell
pip install init-provider
```
Using `uv`:
```shell
uv add init-provider
```
## 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 from `BaseProvider`
Create a class that inherits from `BaseProvider`. This automatically
registers your provider inside the framework.
```python
from init_provider import BaseProvider
class WeatherProvider(BaseProvider):
"""Fetch weather data from the API."""
```
### Store in class variables
Use class variables just like you would define fields in a `dataclass`.
```python
# ...
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 using `__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 `__init__()` method.
For example, you might want to initialize a reusable [aiohttp][1] session
during runtime, when the asyncio event loop is already running.
```python
# ...
import asyncio
from aiohttp import ClientSession
class WeatherProvider(BaseProvider):
# ...
_session: ClientSession
def __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 `__init__()`.
Trying to access the `_session` object will trigger the initialization chain.
*Note 2*: The `__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
### Dispose using `__del__()`
If you need to, you can dispose of resources in the `__del__()` method.
An example of this is closing a database connection.
### Decorate methods using `@init`
Providers are great for encapsulating reusable business logic in a methods.
Every provider method decorated with `@init` becomes guarded. 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.
```python
from init_provider import BaseProvider
class WeatherProvider(BaseProvider):
# ...
@init
def get_url(cls, path: str) -> str:
return f"{cls._base_url}/{path}"
```
### Dependencies using `@requires`
Use the `@requires` decorator to list other providers that the
`WeatherProvider` depends on.
```python
@requires(GeoProvider)
class WeatherProvider(BaseProvider):
# ...
```
## Examples
### Weather service
```python
import asyncio
import logging
from aiohttp import ClientSession
from init_provider import BaseProvider, init, requires
logging.basicConfig(level=logging.DEBUG, format="%(levelname)-8s %(message)s")
class GeoService(BaseProvider):
@init
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 __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()
@init
async def temperature(self, city: str) -> float:
lat, lon = GeoService.city_coordinates(city)
params: dict[str, str | float] = {
"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
# __del__() 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:
```shell
$ 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: 11.40°C
New York: 16.50°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
```python
import logging
import os
import sqlite3
import warnings
from contextlib import contextmanager
from typing import Generator
from init_provider import BaseProvider, init, 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 __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 __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 __del__(self):
os.unlink(self.db_path)
# ↓ Any call to the `conn` method will cause the
# provider to be initialized, if not already done.
@init
@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 __init__() method here! Because there is nothing
# to initialize inside this specific provider itself.
# ↓ Require initialization of all dependencies when this
# method is called.
@init
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:
```shell
$ 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`:
```python
import logging
logging.basicConfig(level=logging.DEBUG)
```
Which will allow you to see what `init_provider` is doing:
```shell
$ 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...
```
## Design choices
* `BaseProvider` inheritance instead of `@provider` decorator.
If a decorator approach was chosen to mimic the `@dataclass` approach,
then the dependencies would have to be specified using another
decorator (ugly) or inside the class body (awkward).
That's why the choice was made towards inheritance. It is very familiar
to `pydantic` users and allows the list of dependencies to be specified
elegantly using the `@requires` decorator sitting on top of the class
definition.
* `__init__()` as initialization method instead of `provider_init()`.
The choice was made use native `__init__()`, because that's where any
developer would expect to find the initialization code. Introducing
a separate `provider_init()` method would inflict additional cognitive
load for no real benefit.
There could be one justification for custom methods: async
initialization. In that case, two base classes would exist:
`BaseProvider` and `AsyncBaseProvider`. Each would define
`provider_init()` and `provider_dispose()` methods as normal functions
or as coroutines. However, the whole async approach had to be discarded
because of incompatibility with guarded class variables (see below).
* `__del__()` as disposal method.
While the `__del__()` method is rarely used and is unfamiliar to most
Python developers, it is the least awkard way to define a sort of a
destructor for a provider.
Even though the **true** implementation of `__del__()` explicitly
states the it is guaranteed to be called, it does not affect the
provider, because we hijack both the `__init__` and `__del__` and they
are guaranteed to be called.
An alternative would be to force the developers to deal with
restrictive `weakref.finalize` functions which are hard to use because
you can't pass a reference to the instance you want to dispose of
as an argument to the finalizer.
* `@init`
While technically the `@init` decorator for guarded class methods is
not required at all, it has been added for type checking purposes.
Without `@init`, we could forcibly convert all methods of the
class into `classmethod`s, adding dependency initialization checks
around them and allowing them to be called without an instance.
```python
"""Without @init but with failing type checks"""
class MyProvider(BaseProvider):
def greet(self, name: str):
print(f"Hello, {name}!")
if __name__ == "__main__":
MyProvider.greet("World")
```
That works and looks very nice, but the static type analyzers go crazy.
They expect the `greet` method to receive 2 arguments, but we only
pass one.
An alternative would be to overhaul the entire provider protocol and
require an actual instance creation. However, under the hood, we
could always return a singleton instance from the `__new__` method.
```python
"""Alternative to @init with instance creation"""
class MyProvider(BaseProvider):
def greet(self, name: str):
print(f"Hello, {name}!")
if __name__ == "__main__":
MyProvider().greet("World")
```
That would work too, but it is so awkward that it's painful to use.
This is not Java and in Python nobody just creates instances left and
right for no good reason. Any Python developer that reads the code
above would immediately assume that a one-time instance of `MyProvider`
was created and would be destroyed soon, which seems counter
intuitive.
There is, however, still one reason for `MyProvider()` approach
to remain on the table: **AI coding assistant and tab-completion.**
During tests, most of the time, when not surrounded by examples or
without explicit instructions, they attempt to write
`MyProvider().greet("World")` instead of `MyProvider.greet("World")`.
If this will remain the case, **the library might eventually switch
to this awkward but practical approach.**
That's why we have the `@init` decorator. It's not there to make
providers and guarded methods work. It is there to solve the type
checking while being the least awkward approach.
* No `async` initialization or disposal.
There are use cases where we'd want both the `__init__` and
`__dispose__` to be async and the whole initialization and disposal of
a particular provider to happen inside an already running event loop.
For example, an async http client:
```python
import asyncio
import aiohttp
class MyProvider(BaseProvider):
client: aiohttp.ClientSession
async def __init__(self):
self.client = aiohttp.ClientSession()
async def __dispose__(self):
await self.client.close()
```
That would look and feel amazing. We could impose a rule that async
providers can depend on sync and other async providers. But the sync
providers would only be allowed to depend on other sync providers,
which would address most concerns around async initialization order.
Unfortunately, this whole logic breaks down the moment we attempt to
implement guarded class variables.
Imagine a function which accesses the `client` class variable above
for some purposes:
```python
async def main():
await MyProvider.client.get("https://example.com")
if __name__ == "__main__":
asyncio.run(main())
```
As soon as `MyProvider.client` code is reached, the provider library
must detect that an attempt to read a guarded class variable is made.
It checks whether the provider is initialized or not. If it is not,
it initializes all of the dependencies of this provider and the
provider itself. Once that's done, the `client` value, already
initialized at this point, is returned.
The only way to implement this detection and lazy initialization
at access time is to override the `__getattribute__` method.
In a situation where provider's initialization logic is asynchronous
the provider chain initialization funciton would have to be called
with `await`. But that would be impossible inside a normal function
or method, such as `__getattribute__`.
Therefore, a tradeoff must be made. Either we want guarded class
variables that are lazily initialized at runtime. Or we want
providers that can have asynchronous initialization logic.
Can't have both.
In this library, the choice was made in favor of the former. Lazily
initialized guarded class variables are too convenient to give them
up.
## License
Licensed under the [Apache-2.0 License](./LICENSE).
[1]: https://docs.aiohttp.org/
[2]: https://martinfowler.com/eaaCatalog/repository.html
Raw data
{
"_id": null,
"home_page": null,
"name": "init-provider",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": "vduseev <vagiz@duseev.com>",
"keywords": "dependency, initialization, lifecycle, provider, singleton",
"author": null,
"author_email": "vduseev <vagiz@duseev.com>",
"download_url": "https://files.pythonhosted.org/packages/8a/cb/dec78e80d7d5863067b4d2fde49f82652e190f19fc96ddd89dbe581a4b1e/init_provider-0.1.3.tar.gz",
"platform": null,
"description": "<h1><code>init-provider</code></h1>\n\nInitialization and instance provider framework for Python.\n\n\n\n\n\n\n* **Init order**: for example, `ProviderA` depends on `ProviderB`.\n* **Reusable objects**: expose instances of Settings or a Connection Pool.\n* **Business logic**: clean internal APIs.\n* **Entry point**: use for a CLI, Web API, background worker, etc.\n\n- [Quick start](#quick-start)\n- [Installation](#installation)\n- [Usage](#usage)\n - [Inherit from `BaseProvider`](#inherit-from-baseprovider)\n - [Store in class variables](#store-in-class-variables)\n - [Initialize using `__init__()`](#initialize-using-__init__)\n - [Dispose using `__del__()`](#dispose-using-__del__)\n - [Decorate methods using `@init`](#decorate-methods-using-init)\n - [Dependencies using `@requires`](#dependencies-using-requires)\n- [Examples](#examples)\n - [Weather service](#weather-service)\n - [User service](#user-service)\n- [Troubleshooting](#troubleshooting)\n - [Enable logging](#enable-logging)\n- [Design choices](#design-choices)\n- [License](#license)\n\n## Quick start\n\nBelow is a full runnable example that uses everything in this library.\n\n```python\nimport logging\nfrom pathlib import Path\nfrom init_provider import BaseProvider, init, requires, setup, dispose\n\n\n@setup\ndef configure() -> None:\n logging.basicConfig(\n level=logging.DEBUG,\n format=\"%(levelname)-8s %(name)-15s %(message)s\",\n )\n if not Path(\"file.txt\").exists():\n logging.info(\"> file.txt does not yet exist\")\n\n\n@dispose\ndef cleanup() -> None:\n if not Path(\"file.txt\").exists():\n logging.info(\"> file.txt no longer exist\")\n\n\nclass Storage(BaseProvider):\n path = Path(\"file.txt\")\n\n def __init__(self):\n logging.info(\"> create Storage\")\n self.path.touch()\n\n def __del__(self):\n logging.info(\"> dispose of Storage\")\n self.path.unlink()\n\n @init\n def write(self, content: str) -> None:\n logging.info(f\"> write to Storage: {content}\")\n self.path.write_text(content)\n\n @init\n def read(self) -> str:\n data = self.path.read_text()\n logging.info(f\"> read from Storage: {data}\")\n return data\n\n\n@requires(Storage)\nclass Namer(BaseProvider):\n def __init__(self) -> None:\n logging.info(\"> create Namer\")\n Storage.write(\"Bobby\")\n\n\n@requires(Namer)\nclass Greeter(BaseProvider):\n define_at_runtime: str\n\n def __init__(self):\n logging.info(\"> create Greeter\")\n self.define_at_runtime = Storage.read()\n\n @init\n def greet(self) -> None:\n print(f\">>> Hello, {self.define_at_runtime}!\")\n\n\nif __name__ == \"__main__\":\n Greeter.greet()\n```\n\nOutput:\n\n```shell\n$ uv run python examples/full_example.py\nINFO root > file.txt does not yet exist\nINFO init_provider Setup hook executed.\nDEBUG init_provider About to initialize provider Greeter because of: greet\nDEBUG init_provider Initialization order for provider Greeter is: Storage, Namer, Greeter\nDEBUG init_provider Initializing provider Storage...\nINFO root > create Storage\nINFO init_provider Provider Storage initialized\nDEBUG init_provider Initializing provider Namer...\nINFO root > create Namer\nINFO root > write to Storage: Bobby\nINFO init_provider Provider Namer initialized\nDEBUG init_provider Initializing provider Greeter...\nINFO root > create Greeter\nINFO root > read from Storage: Bobby\nINFO init_provider Provider Greeter initialized\n>>> Hello, Bobby!\nDEBUG init_provider Provider dispose call order: ['Greeter', 'Namer', 'Storage']\nINFO init_provider Dispose hook for Greeter was executed.\nINFO init_provider Dispose hook for Namer was executed.\nINFO root > dispose of Storage\nINFO init_provider Dispose hook for Storage was executed.\nINFO root > file.txt no longer exist\nINFO init_provider Dispose hook executed.\n```\n\n## Installation\n\nUsing `pip`:\n\n```shell\npip install init-provider\n```\n\nUsing `uv`:\n\n```shell\nuv add init-provider\n```\n\n## Usage\n\nProviders are just classes. In fact, they look a lot like a `dataclass` but\nwith three major differences:\n\n1. You do not need to instantiate the provider class.\n2. Providers can depend on each other.\n3. Calling any method or attribute of a provider will trigger initialization.\n\n### Inherit from `BaseProvider`\n\nCreate a class that inherits from `BaseProvider`. This automatically\nregisters your provider inside the framework. \n\n```python\nfrom init_provider import BaseProvider\n\nclass WeatherProvider(BaseProvider):\n \"\"\"Fetch weather data from the API.\"\"\"\n```\n\n### Store in class variables\n\nUse class variables just like you would define fields in a `dataclass`.\n\n```python\n# ...\nclass WeatherProvider(BaseProvider):\n # ...\n _base_url: str = \"https://theweather.com/api\"\n```\n\n*Note*: `init_provider` doesn't care about underscores in variable and\nmethod names. It will expose them all the same.\n\n### Initialize using `__init__()`\n\nWhen you need to initialize the provider, you can focus on **what** needs to\nbe initialized rather than **when** it needs to be initialized.\n\nNot all providers require initialization, but when they do, you can define\nit inside the `__init__()` method.\n\nFor example, you might want to initialize a reusable [aiohttp][1] session\nduring runtime, when the asyncio event loop is already running.\n\n```python\n# ...\nimport asyncio\nfrom aiohttp import ClientSession\n\nclass WeatherProvider(BaseProvider):\n # ...\n _session: ClientSession\n\n def __init__(self) -> None:\n self._session = ClientSession()\n\nif __name__ == \"__main__\":\n if WeatherProvider._session.closed:\n print(\"Session is still closed\")\n```\n\n*Note 1*: in the example aboev, the `_session` variable is declared without\na value. The initialization is done inside the `__init__()`.\nTrying to access the `_session` object will trigger the initialization chain.\n\n*Note 2*: The `__init__` method of the owner class is the only place\nwhere initialization will not be triggered, when the object is accessed.\n\n*Warning*: Declaring a class variable with a default value will mean that it's\n\n### Dispose using `__del__()`\n\nIf you need to, you can dispose of resources in the `__del__()` method.\nAn example of this is closing a database connection.\n\n### Decorate methods using `@init`\n\nProviders are great for encapsulating reusable business logic in a methods.\nEvery provider method decorated with `@init` becomes guarded. Guarded\nmethods cause initialization of the provider chain when they are called.\n\n*Note*: Reserved methods that contain double underscore (`__`) and methods\ndecorated with `@staticmethod` or `@classmethod` will not be guarded.\n\n```python\nfrom init_provider import BaseProvider\n\nclass WeatherProvider(BaseProvider):\n # ...\n @init\n def get_url(cls, path: str) -> str:\n return f\"{cls._base_url}/{path}\"\n```\n\n### Dependencies using `@requires`\n\nUse the `@requires` decorator to list other providers that the\n`WeatherProvider` depends on.\n\n```python\n@requires(GeoProvider)\nclass WeatherProvider(BaseProvider):\n # ...\n```\n\n## Examples\n\n### Weather service\n\n```python\nimport asyncio\nimport logging\nfrom aiohttp import ClientSession\nfrom init_provider import BaseProvider, init, requires\n\nlogging.basicConfig(level=logging.DEBUG, format=\"%(levelname)-8s %(message)s\")\n\n\nclass GeoService(BaseProvider):\n @init\n def city_coordinates(self, name: str) -> tuple[float, float]:\n \"\"\"Returns the latitude and longitude of a city.\"\"\"\n if name == \"London\":\n return 51.509, -0.118 # London, UK\n elif name == \"New York\":\n return 40.7128, -74.0060 # New York, USA\n raise ValueError(f\"Unknown city: {name}\")\n\n\n@requires(GeoService)\nclass WeatherService(BaseProvider):\n _session: ClientSession\n _base_url: str = \"https://api.open-meteo.com/v1/forecast/\"\n\n def __init__(self) -> None:\n # Properly initializing aiohttp session at runtime, when the\n # default asyncio loop is already running.\n self._session = ClientSession(self._base_url)\n\n @classmethod\n async def close(cls):\n await cls._session.close()\n\n @init\n async def temperature(self, city: str) -> float:\n lat, lon = GeoService.city_coordinates(city)\n params: dict[str, str | float] = {\n \"latitude\": lat,\n \"longitude\": lon,\n \"hourly\": \"temperature_2m\",\n }\n async with self._session.get(self._base_url, params=params) as resp:\n data = await resp.json()\n return data[\"hourly\"][\"temperature_2m\"][0]\n\n\nasync def main():\n # This will immediately initialize WeatherService and its dependencies,\n # because we have attempted to access the _session property.\n print(f\"Is session closed: {WeatherService._session.closed}\")\n\n # Subsequent calls do not reinitialize the provider.\n london = await WeatherService.temperature(\"London\")\n new_york = await WeatherService.temperature(\"New York\")\n print(f\"London: {london:.2f}\u00b0C\")\n print(f\"New York: {new_york:.2f}\u00b0C\")\n\n # Release the resources. Normally, this would be implemented in the\n # __del__() method of the provider, but the async client must be closed\n # inside the same event loop it was created.\n await WeatherService.close()\n print(f\"Is session closed: {WeatherService._session.closed}\")\n\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\nOutput:\n\n```shell\n$ uv run python examples/weather_service.py\nDEBUG Using selector: KqueueSelector\nDEBUG About to initialize provider WeatherService because of: _session\nDEBUG Initialization order for provider WeatherService is: GeoService, WeatherService\nDEBUG Initializing provider GeoService...\nINFO Provider GeoService initialized\nDEBUG Initializing provider WeatherService...\nINFO Provider WeatherService initialized\nIs session closed: False\nLondon: 11.40\u00b0C\nNew York: 16.50\u00b0C\nIs session closed: True\nDEBUG Provider dispose call order: ['WeatherService', 'GeoService']\nINFO Dispose hook for WeatherService was executed.\nINFO Dispose hook for GeoService was executed.\n```\n\n### User service \n\n```python\nimport logging\nimport os\nimport sqlite3\nimport warnings\nfrom contextlib import contextmanager\nfrom typing import Generator\n\nfrom init_provider import BaseProvider, init, requires, setup\n\n\n# (Optional) Declare a setup function to be executed once per application\n# process before any provider is initialized.\n@setup\ndef configure():\n log_format = \"%(levelname)-8s %(message)s\"\n logging.basicConfig(level=logging.DEBUG, format=log_format)\n warnings.filterwarnings(\"ignore\", module=\"some_module\")\n\n\n# \u2193 Basic provider. Exposes 1 attribute: connection\nclass DatabaseService(BaseProvider):\n \"\"\"Single instance of connection ot SQLite.\"\"\"\n\n # \u2193 Any attempt to access a provider attribute outside\n # of __init__() will cause the provider to be initialized.\n db_path: str\n\n # \u2193 Initialize, just like in a dataclass. But you NEVER\n # have to create an instance of a provider manually.\n def __init__(self) -> None:\n # Run some one-time initialization logic\n self.db_path = \"database.db\"\n\n # Initialize the database. This will only be done once\n # across the entire lifecycle of the application.\n with sqlite3.connect(self.db_path) as conn:\n cur = conn.cursor()\n # Create a table\n cur.execute(\n \"CREATE TABLE IF NOT EXISTS users \"\n \"(id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT)\"\n )\n # Add mock data\n cur.executemany(\n \"INSERT INTO users (name) VALUES (?)\",\n [(\"Alice\",), (\"Bob\",)],\n )\n conn.commit()\n\n # \u2193 Declare a dispose method to be called before the application exits.\n def __del__(self):\n os.unlink(self.db_path)\n\n # \u2193 Any call to the `conn` method will cause the\n # provider to be initialized, if not already done.\n @init\n @contextmanager\n def conn(self) -> Generator[sqlite3.Connection, None, None]:\n \"\"\"One-time connection to the database.\"\"\"\n with sqlite3.connect(self.db_path) as conn:\n yield conn\n\n\n# \u2193 This one depends on another provider.\n@requires(DatabaseService)\nclass UserService(BaseProvider):\n \"\"\"Intenal API class to abstract the Users data layer.\"\"\"\n\n # \u2192 Notice: NO __init__() method here! Because there is nothing\n # to initialize inside this specific provider itself.\n\n # \u2193 Require initialization of all dependencies when this\n # method is called.\n @init\n def get_name(self, user_id: int) -> str | None:\n \"\"\"Get user name based on ID\"\"\"\n\n # \u2193 Access the method from another provider\n with DatabaseService.conn() as conn:\n cur = conn.cursor()\n if result := cur.execute(\n \"SELECT name FROM users WHERE id = ?\", (user_id,)\n ).fetchone():\n return result[0]\n else:\n return None\n\n\nif __name__ == \"__main__\":\n # \u2193 This will cause the chain of dependencies to be\n # initialized in the following order:\n # 1. configure() function will be called\n # 2. DatabaseService\n database_path = DatabaseService.db_path\n print(f\">> {database_path}\")\n\n # \u2193 This will only initialize the UserService, because\n # its dependencies are already initialized.\n user_1 = UserService.get_name(1)\n print(f\">> {user_1}\")\n\n # \u2193 Let's get the name of another user. NOTHING extra will be\n # done because the dependency graph is already initialized.\n user_2 = UserService.get_name(2)\n print(f\">> {user_2}\")\n```\n\nOutput:\n\n```shell\n$ uv run python examples/user_service.py\nINFO Setup hook executed.\nDEBUG About to initialize provider DatabaseService because of: db_path\nDEBUG Initialization order for provider DatabaseService is: DatabaseService\nDEBUG Initializing provider DatabaseService...\nINFO Provider DatabaseService initialized\nDEBUG About to initialize provider UserService because of: get_name\nDEBUG Initialization order for provider UserService is: DatabaseService (initialized), UserService\nDEBUG Initializing provider UserService...\nINFO Provider UserService initialized\n>> database.db\n>> Alice\n>> Bob\nDEBUG Provider dispose call order: ['UserService', 'DatabaseService']\nINFO Dispose hook for UserService was executed.\nINFO Dispose hook for DatabaseService was executed.\n```\n\n## Troubleshooting\n\n### Enable logging\n\nThe framework produces logs tied to the `init_provider` module. Make sure\nthe logs from this module are not suppressed in the global `logging`\nconfiguration.\n\nThe easiest way to enable logging is to set the logging level to `DEBUG`:\n\n```python\nimport logging\nlogging.basicConfig(level=logging.DEBUG)\n```\n\nWhich will allow you to see what `init_provider` is doing:\n\n```shell\n$ uv run python examples/weather_service.py\nDEBUG About to initialize provider WeatherService because of: session\nDEBUG Initialization order for provider WeatherService is: GeoService, WeatherService\nDEBUG Initializing provider GeoService...\nINFO Provider GeoService initialized successfully\nDEBUG Initializing provider WeatherService...\n```\n\n## Design choices\n\n* `BaseProvider` inheritance instead of `@provider` decorator.\n\n If a decorator approach was chosen to mimic the `@dataclass` approach,\n then the dependencies would have to be specified using another\n decorator (ugly) or inside the class body (awkward).\n\n That's why the choice was made towards inheritance. It is very familiar\n to `pydantic` users and allows the list of dependencies to be specified\n elegantly using the `@requires` decorator sitting on top of the class\n definition.\n\n* `__init__()` as initialization method instead of `provider_init()`.\n\n The choice was made use native `__init__()`, because that's where any\n developer would expect to find the initialization code. Introducing\n a separate `provider_init()` method would inflict additional cognitive\n load for no real benefit.\n\n There could be one justification for custom methods: async\n initialization. In that case, two base classes would exist:\n `BaseProvider` and `AsyncBaseProvider`. Each would define\n `provider_init()` and `provider_dispose()` methods as normal functions\n or as coroutines. However, the whole async approach had to be discarded\n because of incompatibility with guarded class variables (see below).\n\n* `__del__()` as disposal method.\n\n While the `__del__()` method is rarely used and is unfamiliar to most\n Python developers, it is the least awkard way to define a sort of a \n destructor for a provider.\n\n Even though the **true** implementation of `__del__()` explicitly\n states the it is guaranteed to be called, it does not affect the\n provider, because we hijack both the `__init__` and `__del__` and they\n are guaranteed to be called.\n\n An alternative would be to force the developers to deal with\n restrictive `weakref.finalize` functions which are hard to use because\n you can't pass a reference to the instance you want to dispose of\n as an argument to the finalizer.\n\n* `@init`\n\n While technically the `@init` decorator for guarded class methods is\n not required at all, it has been added for type checking purposes.\n\n Without `@init`, we could forcibly convert all methods of the\n class into `classmethod`s, adding dependency initialization checks\n around them and allowing them to be called without an instance.\n\n ```python\n \"\"\"Without @init but with failing type checks\"\"\"\n\n class MyProvider(BaseProvider):\n def greet(self, name: str):\n print(f\"Hello, {name}!\")\n\n if __name__ == \"__main__\":\n MyProvider.greet(\"World\")\n ```\n\n That works and looks very nice, but the static type analyzers go crazy.\n They expect the `greet` method to receive 2 arguments, but we only\n pass one.\n\n An alternative would be to overhaul the entire provider protocol and\n require an actual instance creation. However, under the hood, we\n could always return a singleton instance from the `__new__` method.\n\n ```python\n \"\"\"Alternative to @init with instance creation\"\"\"\n\n class MyProvider(BaseProvider):\n def greet(self, name: str):\n print(f\"Hello, {name}!\")\n\n if __name__ == \"__main__\":\n MyProvider().greet(\"World\")\n ```\n\n That would work too, but it is so awkward that it's painful to use.\n This is not Java and in Python nobody just creates instances left and\n right for no good reason. Any Python developer that reads the code \n above would immediately assume that a one-time instance of `MyProvider`\n was created and would be destroyed soon, which seems counter \n intuitive.\n\n There is, however, still one reason for `MyProvider()` approach\n to remain on the table: **AI coding assistant and tab-completion.**\n During tests, most of the time, when not surrounded by examples or\n without explicit instructions, they attempt to write\n `MyProvider().greet(\"World\")` instead of `MyProvider.greet(\"World\")`.\n If this will remain the case, **the library might eventually switch\n to this awkward but practical approach.**\n\n That's why we have the `@init` decorator. It's not there to make\n providers and guarded methods work. It is there to solve the type\n checking while being the least awkward approach.\n\n* No `async` initialization or disposal.\n\n There are use cases where we'd want both the `__init__` and\n `__dispose__` to be async and the whole initialization and disposal of\n a particular provider to happen inside an already running event loop.\n\n For example, an async http client:\n\n ```python\n import asyncio\n import aiohttp\n\n class MyProvider(BaseProvider):\n client: aiohttp.ClientSession\n\n async def __init__(self):\n self.client = aiohttp.ClientSession()\n\n async def __dispose__(self):\n await self.client.close()\n ```\n\n That would look and feel amazing. We could impose a rule that async\n providers can depend on sync and other async providers. But the sync\n providers would only be allowed to depend on other sync providers,\n which would address most concerns around async initialization order.\n\n Unfortunately, this whole logic breaks down the moment we attempt to\n implement guarded class variables.\n\n Imagine a function which accesses the `client` class variable above\n for some purposes:\n\n ```python\n async def main():\n await MyProvider.client.get(\"https://example.com\")\n\n if __name__ == \"__main__\":\n asyncio.run(main())\n ```\n\n As soon as `MyProvider.client` code is reached, the provider library\n must detect that an attempt to read a guarded class variable is made.\n It checks whether the provider is initialized or not. If it is not,\n it initializes all of the dependencies of this provider and the\n provider itself. Once that's done, the `client` value, already\n initialized at this point, is returned.\n\n The only way to implement this detection and lazy initialization\n at access time is to override the `__getattribute__` method.\n \n In a situation where provider's initialization logic is asynchronous\n the provider chain initialization funciton would have to be called\n with `await`. But that would be impossible inside a normal function\n or method, such as `__getattribute__`.\n\n Therefore, a tradeoff must be made. Either we want guarded class\n variables that are lazily initialized at runtime. Or we want\n providers that can have asynchronous initialization logic.\n Can't have both.\n\n In this library, the choice was made in favor of the former. Lazily\n initialized guarded class variables are too convenient to give them\n up.\n\n## License\n\nLicensed under the [Apache-2.0 License](./LICENSE).\n\n[1]: https://docs.aiohttp.org/\n[2]: https://martinfowler.com/eaaCatalog/repository.html",
"bugtrack_url": null,
"license": null,
"summary": null,
"version": "0.1.3",
"project_urls": {
"Documentation": "https://github.com/vduseev/init-provider",
"Homepage": "https://github.com/vduseev/init-provider",
"Issues": "https://github.com/vduseev/init-provider/issues",
"Repository": "https://github.com/vduseev/init-provider"
},
"split_keywords": [
"dependency",
" initialization",
" lifecycle",
" provider",
" singleton"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "c2047a38aefddc30b82f2c1192b938cc13f1777b7624f8bef33225a94e3b3263",
"md5": "996124287e04fa0442464d43f5e0cab8",
"sha256": "f9792fe13a96352d0e1085e3da8a67755121f2b47b84e44a67058a5831720b44"
},
"downloads": -1,
"filename": "init_provider-0.1.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "996124287e04fa0442464d43f5e0cab8",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 24742,
"upload_time": "2025-10-12T13:35:32",
"upload_time_iso_8601": "2025-10-12T13:35:32.030704Z",
"url": "https://files.pythonhosted.org/packages/c2/04/7a38aefddc30b82f2c1192b938cc13f1777b7624f8bef33225a94e3b3263/init_provider-0.1.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "8acbdec78e80d7d5863067b4d2fde49f82652e190f19fc96ddd89dbe581a4b1e",
"md5": "7c42b8b6793f20d130e5c0247b2b3e15",
"sha256": "a3cd8d85ad560a2bbb05eafed017b30df05189001d5e369556b8a76ff2bfd821"
},
"downloads": -1,
"filename": "init_provider-0.1.3.tar.gz",
"has_sig": false,
"md5_digest": "7c42b8b6793f20d130e5c0247b2b3e15",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 31492,
"upload_time": "2025-10-12T13:35:33",
"upload_time_iso_8601": "2025-10-12T13:35:33.694655Z",
"url": "https://files.pythonhosted.org/packages/8a/cb/dec78e80d7d5863067b4d2fde49f82652e190f19fc96ddd89dbe581a4b1e/init_provider-0.1.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-10-12 13:35:33",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "vduseev",
"github_project": "init-provider",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "init-provider"
}
JSON
