Why do I get "Error, database connection not opened" in my FastAPI & peewee application?

Question

I'm new to Python and I'm working on a FastAPI & peewee application. I want to manage the database connection pool explicitly so according to the framework integration documentation for FastAPI I use the startup/shutdown events.

database = PooledSqliteDatabase('foobar.db', autoconnect=False, stale_timeout=60)

with database:
  database.create_tables([models.Foobar])

app = FastAPI()

app.include_router(foobar_router)

@app.on_event("startup")
def startup():
  database.connect()

@app.on_event("shutdown")
def shutdown():
  if not database.is_closed():
    database.close()

The problem is that these events do not correspond to "open the connection when a request is received, then close it when the response is returned", but when the whole app starts and stops. More importantly I don't understand why I get "Error, database connection not opened." when I access the database in my router as it should still be opened at that point? It only seems to work when I add async to my routes.

I guess I could switch to some proper "before/after request" middleware, but at this point I'm wondering if my approach is just completely wrong.

Edit: I tried using middleware, but while this unsurprisingly works as intended, I still have the issue that I get "Error, database connection not opened." unless I add async to my routes.

@app.middleware("http")
async def with_database(request: Request, call_next):
  with database:
    response = await call_next(request)
  return response

Answer 1

Note your pool probably won't work as expected in an asyncio environment. Peewee's connection model, including the pool, is built around a thread-per-connection. With async the connections are pooled, but since everything is running in the same thread, all your coroutines will be sharing the same connection. This doesn't present any problems for standard threaded or gevent code (which patches thread local to be green-thread local).

That said, I don't use FastAPI but extrapolating from their (insane) sqlalchemy doc, you might try the following approach if you want to avoid a per-request middleware:

from fastapi import Depends, FastAPI, HTTPException


app = FastAPI()

database = PooledSqliteDatabase('foobar.db', autoconnect=False, stale_timeout=60)

with database:
    database.create_tables([models.Foobar])

def ensure_connection():
    database.connect()
    try:
        yield database
    finally:
        database.close()

@app.post("/users/", ...)
def create_user(..., database=Depends(ensure_connection)):
    ...

I think a per-request middleware is probably cleaner to implement, however. Given my caveat at the beginning of the comment, I'd strongly suggest you analyze your application and verify that it's using connections properly -- very likely (and this will depend on when you yield to the event loop) it is not, though.