trainspotting
README.md
Choose dependency injection
- Friendly with MyPy
- Supports lazy injections
- Supports fabrics with environment variables
- Can connect/disconnect clients in correct order
Install
pip install trainspotting
Contact
Feel free to ask questions in telegram t.me/PulitzerKenny
Examples
from typing import Protocolfrom dataclasses import dataclassfrom trainspotting import DependencyInjectorfrom trainspotting.factory import factory, EnvField
class TransportProtocol(Protocol): def go(self): ...
class EngineProtocol(Protocol): def start(self): ...
class HeadlightsProtocol(Protocol): def light(self): ...
@dataclassclass Driver: transport: TransportProtocol def drive(self): self.transport.go()
@dataclassclass V8Engine: sound: str
def start(self): print(self.sound)
@dataclassclass XenonHeadlights: brightness: int
def light(self): print(f'LIGHT! {self.brightness}')
@dataclassclass Car: engine: EngineProtocol headlights: HeadlightsProtocol
def go(self): self.engine.start() self.headlights.light()
def get_config(): return { EngineProtocol: factory(V8Engine, sound=EnvField('ENGINE_SOUND')), HeadlightsProtocol: factory( XenonHeadlights, brightness=EnvField('HEADLIGHTS_BRIGHTNESS', int) ), TransportProtocol: Car, }
injector = DependencyInjector(get_config())injected_driver = injector.inject(Driver)injected_driver().drive()
Clients connect/disconnect
Connect can be used for async class initialization
import aioredis
from typing import Protocolfrom trainspotting import DependencyInjector
class ClientProtocol(Protocol): async def do(self): ...
class RedisClient: def __init__(self): self.pool = None async def do(self): if not self.pool: raise ValueError print('did something') async def connect(self): self.pool = await aioredis.create_redis_pool(('127.0.0.1', 6379)) print('connected') async def disconnect(self): self.pool.close() await self.pool.wait_closed() print('disconnected')
async def main(client: ClientProtocol): await client.do()
injector = DependencyInjector({ClientProtocol: RedisClient})injected = injector.inject(main)
async with injector.deps: # connected await injected() # did something# disconnected
Types Injection
class EntityProtocol(Protocol): def do(self): ...
class Entity: def do(self): print('do something')
@dataclassclass SomeUsefulClass: entity: Type[EntityProtocol]
injector.add_config({EntityProtocol: Entity})injector.inject(SomeUsefulClass)
Lazy injections
@injector.lazy_injectasync def some_func(client: ClientProtocol): await client.do_something()
some_func() # raise TypeError, missing argument clientinjector.do_lazy_injections()some_func() # ok
Extend or change config
injector.add_config({ClientProtocol: Client})
EnvField
import osfrom typing import Protocolfrom dataclasses import dataclassfrom trainspotting import DependencyInjectorfrom trainspotting.factory import factory, EnvField
class ClientProtocol(Protocol): def do(self): ...
@dataclassclass Client(ClientProtocol): url: str log_prefix: str timeout: int = 5
def do(self): print(f'{self.log_prefix}: sent request to {self.url} with timeout {self.timeout}')
injector = DependencyInjector({ ClientProtocol: factory( Client, url=EnvField('SERVICE_URL'), log_prefix=EnvField('LOG_PREFIX', default='CLIENT'), timeout=EnvField('SERVICE_TIMEOUT', int, optional=True), )})
def main(client: ClientProtocol): client.do() os.environ.update({'SERVICE_URL': 'some_url'})injected = injector.inject(main)injected() # CLIENT: sent request to some_url with timeout 5
Supports type validation
from typing import Protocol, runtime_checkablefrom trainspotting import DependencyInjector
@runtime_checkableclass ClientProtocol(Protocol): def do(self): ...
class Client: def do(self): print('did something')
class BadClient: def wrong_do(self): ...
def main(client: ClientProtocol): client.do()
injector = DependencyInjector({ ClientProtocol: BadClient,})
injector.inject(main)() # raise InjectionError: <class 'BadClient'> does not realize <class 'ClientProtocol'> interface
injector = DependencyInjector({ ClientProtocol: Client,})
injector.inject(main)() # prints: did something
License
MIT license. Copyright (c) 2022 Avito
Описание
Python Dependency Injector based on interface binding
Языки
Python
- Makefile