dependencies
dependencies copied to clipboard
Type hints.
from dependencies import wire, scope
class Container:
user = wire(User)
name = 'John'
@wire
async def age() -> int:
yield 1
async with scope(Container) as container:
print(container.user.name)
print(container.user.age)
print(container.user is container.user)
Kate
1
True
class Container:
@wire(User)
def username(user: User) -> str:
return user.name
name = 'Kate'
age = 18
print(Container.username)
Kate
With new syntax I think its possible to keep following features of the library:
- resolve dependencies based on arguments name
- sticky scopes
- values (redo with native
@property) - Setup and teardown
- class-named attributes (redo with
Type[Service]annotation including@propertyreturning Type) - overrides with arguments to call (positional argument
Q('deep.attr.name', value)for deep implementation detail overrides) - attribute access (redo via
Attribute['name'] annotation, maybeoperator.attrgetter` as a value, containers defined below in the code referenced same way regular classes do, no nested injectors are allowed). - lazy evaluation (evaluate declaration on first use, redo via future annotations)
- recursion error tracking
-
thisobject would be reimplemented using access modificator (decorated callback)
New features:
- mypy friendly, works as dataclass alias by default.
- maybe based on decorator instead of inheritance from base class.
- totally ban inheritance
- don't care about *args and **kwargs
Abandoned features:
- Package
- Shield ?!
- Produce ?!
- Pluck ?!
@supadrupa thoughts?
Let's pretend we have such an example
# db/Table
metadata = MetaData()
account_table = Table(
'account', metadata,
Column('id', UUID, primary_key=True),
Column('avatar', String, nullable=False),
schema="accounts"
)
# account/entity
AccountID = NewType("AccountID", UUID)
@dataclass
class Account:
id: AccountID
avatar: str
class AbstractAccountStore(abc.ABC):
@abc.abstractmethod
async def add(self, account: Account):
pass
@abc.abstractmethod
async def find(self, account_id: AccountID) -> Account | None:
pass
# account/serviece
class AccountService:
def __init__(self, account_store: AbstractAccountStore):
self._account_store = account_store
async def add(self, account: Account):
await self._account_store.add(account=account)
async def find(self, account_id: AccountID):
return await self._account_store.find(account_id=account_id)
# db/account_store
class AccountStore(AbstractAccountStore):
def __init__(self, database: Database):
self._database = database
async def add(self, account: Account):
async with self._database as connect:
query = account_table.insert()
await connect.execute(
query=query,
values=asdict(account)
)
async def find(self, account_id: AccountID) -> Account | None:
query = account_table.select().where(account_table.c.id == account_id)
async with self._database as connect:
row = await connect.fetch_one(query=query)
if row is None:
return None
return self._row_as_account(row)
@staticmethod
def _row_as_account(row):
return Account(**dict(row))
# example
account_service = AccountService(
account_store=AccountStore(
database=Database("postgresql://postgresql:postgresql@localhost:5432/postgresql")
)
)
account = Account(
id=AccountID(uuid4()),
avatar="avatar",
)
await account_service.add(account)
res = await account_service.find(account.id)
In that case I would write it like this:
from dependencies import container
@container
class AccountServiceFactory:
account_service: AccountService
account_store: AccountStore
database: Database
url: str = "postgresql://postgresql:postgresql@localhost:5432/postgresql"
account = Account(id=AccountID(uuid4()), avatar="avatar")
async with AccountServiceFactory() as factory:
await factory.account_service.add(account)
response = await factory.account_service.find(account.id)