Neoteroi
bug: factory methods only receive activating type instead of registered type
Given the following, factory methods fail to produce the expected outcome, and instead raise the NotImplementedError exception.
class Dep:
def __init__(_, value: str):
...
TypeA = Annotated[Dep, "type_a"]
TypeB = Annotated[Dep, "type_b"]
class Fetcher:
def __init__(_, dep: Dep):
...
def factory(scope, dep_type) -> Dep:
if dep_type == TypeA:
value = "a"
elif dep_type == TypeB:
value = "b"
else:
raise NotImplementedError(f"Unknown type: {dep_type}")
return Dep(value)
container.add_transient_by_factory(factory, TypeA)
container.add_transient_by_factory(factory, TypeB)
container.add_transient(Fetcher)
fetcher = container.get(Fetcher)
Hi @StummeJ I tried to understand the use case for supporting this scenario, but I don't get the rationale for it. Can you please give a more realistic example?
Sure, we ran into issues when trying to construct a database connection based on an aliased type
ReadOnlyDb = Annotated[Engine, intern("database_read_only")]
ReadWriteDb = Annotated[Engine, intern("database_read_write")]
def database_client_factory(
scope: ActivationScope, activating_type: Type, registered_type: Type
) -> Engine:
...
if registered_type == ReadOnlyDb:
host = config.reader.host
port = config.reader.port
if "postgresql" in config.dialect:
connection_options.update(
{
"isolation_level": "REPEATABLE READ",
"postgresql_readonly": True,
"postgresql_deferrable": True,
}
)
elif registered_type == ReadWriteDb:
host = config.writer.host
port = config.writer.port
...
# Database
self._container.add_singleton_by_factory(database_client_factory, ReadOnlyDb)
self._container.add_singleton_by_factory(database_client_factory, ReadWriteDb)
If you request a type that relies on ReadOnlyDb or ReadWriteDb then you'll receive that type instead of the registered type
Why not something like the following, if you want to reason in terms of Engine in other functions that need it?
ReadOnlyDb = Annotated[Engine, intern("database_read_only")]
ReadWriteDb = Annotated[Engine, intern("database_read_write")]
def read_only_db_engine_factory(
scope: ActivationScope, activating_type: Type
) -> Engine: ... # return instance of ReadOnlyDb
def read_write_db_engine_factory(
scope: ActivationScope, activating_type: Type
) -> Engine: ... # return instance of ReadWriteDb
# Database
if config.read_only_db:
self._container.add_singleton_by_factory(read_only_db_engine_factory)
else:
self._container.add_singleton_by_factory(read_write_db_engine_factory)
I do something similar in the projects where I want to support different kinds of persistence layer (like here)
if settings.db_connection_string:
# Register SQL DB data access services
else:
# Register alternative data access services
The second argument of the register factory functions was never meant to be used in the way you are suggesting, it was always meant to be used when the factory doesn't have an annotated return type.
And if you know that certain functions need specifically the ReadOnlyDb and some other need ReadWriteDb, I would register those types in the container and request them where needed (== not reasoning in terms of Engine in those contexts).
For the database connection there is work prior to and after the conditional that is relevant to both. Having two methods does solve this issue, but it does create duplication in the codebase. Any thoughts with regards to that?