Decoupled packages with FastAPI

[zach-iwx]

Hello,

So I am working on a Python application which an API using the FastAPI framework and a "watcher" component which will handle some asynchronous things on the backend (checking deployment status, health checks, etc). So what I have right now is 3 containers: ApplicationContainer, APIContainer, and WatcherContainer. The basic package structure is as follows:

.
└── virtual_labs
    ├── api
    │   ├── auth
    │   ├── container.py
    │   ├── data_sources
    │   │   ├── enterprise_dw
    │   │   ├── northwind
    │   │   └── webreports
    │   ├── endpoints
    │   └── services
    ├── container.py
    ├── core
    │   ├── clients
    │   ├── models
    │   ├── mutation_executor
    │   ├── schemas
    │   └── settings
    └── watcher
        └── container.py

Basically the reasoning for this is that I have quite a bit of common code that I would like to be shared between the API and the watcher component. The thing I'm struggling with is how to handle the FastAPI create_app() method and adding the router via dependency injection. Can this be achieve or did I just waste the last several hours of my life?

I was able to get it somewhat working but there's some weird side effects. I started getting warnings that my async resources were not being awaited and seeing this exception get thrown RuntimeError: There is no current event loop in thread 'MainThread' What I was envisioning was something like the following for the main entrypoint:

@inject
async def main(entrypoint: Literal['api', 'watcher'], create_app_factory = Provide[ApplicationContainer.api_package.create_app_factory], watcher_coordinator = Provide[ApplicationContainer.watcher_package.coordinator]):
  if entrypoint == 'api':
    return create_app_factory()
  elif entrypoint == 'watcher':
    await watcher_coordinator.run()


if __name__ == "__main__":
    entrypoint = 'api'
    container = ApplicationContainer()
    settings = get_app_settings()
    settings.configure_logging()

    container.config.from_pydantic(settings)
    container.init_resources()
    
    # would load this via argparser
    entrypoint = "api"
    main(entrypoint)

Any ideas on this topic? After I started I realized it was probably a boneheaded move due to all the complexities introduced but it feels like too late to turn back now.