Starlette: Finesse database backends

[tomchristie]

Follow-ons from #366

• Change the interfaces so that fetchone and fetchall return Record instances.
• Cleanup the transaction and connection state management.
• ~Add SQLite support.~ - See #283 instead

[tomchristie]

See http://github.com/encode/databases instead.
Will close this once we push out DatabaseMiddleware in favor of databases.

[tiangolo]

I have a couple of (7) questions about your future plans, I numbered them so that you can just write something like:

1. correct
2. this and that
3. correct

1. As I'm understanding, the idea is to remove (or stop suggesting): https://www.starlette.io/database/ and instead of that have docs for integrating Starlette with https://github.com/encode/databases, right?

Currently, the API looks something like (shortened):

@app.route("/notes", methods=["GET"])
async def list_notes(request):
    query = notes.select()
    results = await request.database.fetchall(query)
    return JSONResponse(content)

1. In this scenario (the current state of Starlette), connect() is called automatically before starting the request and close()/disconnect() is called automatically after the request (all done by the DatabaseMiddleware), having a DB session per request that lives for just that request. Is that right?

After moving to databases, I imagine the API would probably look something similar to:

from databases import Database

database = Database('postgresql://localhost/example')

@app.route("/notes", methods=["GET"])
async def list_notes(request):

    await database.connect()

    query = notes.select()
    results = await database.fetchall(query)

    await database.disconnect()

    return JSONResponse(content)

1. Is this style of API/usage what you expect to happen then? Or am I completely misunderstanding the plan?

2. Assuming the API looks like this example, in this case, the database is not attached to the request by Starlette, but is used directly, by hand (in code). And .connect() and .disconnect() are called explicitly, to have a single connection/session for this specific request, instead of being done automatically by the middleware. Is that right?

---

I'm asking all this because I want to align what I'm doing in FastAPI with what you are doing upstream (Starlette, databases), as I want to diverge as little as possible and keep as compatible/equivalent to Starlette as possible.

---

On the other side, I want to have docs of how to use "standard" SQLAlchemy ORM models in FastAPI (and I can put them here too if that's interesting).

1. I understand I can't use them with the current implementation of Starlette DBs utilities nor with databases, as SQLAlchemy doesn't have async support directly. And to have ORM support with async I would have to use a different ORM, right?

2. So, to be able to use "standard" SQLAlchemy ORM models (without risking locking) I have to run them with run_in_threadpool() (that happens automatically with the functions are def instead of async def), and then I have to manually create the session in each requests, and manually .close() it at the end of the function, and that would be the version of non-async-standard-SQLAlchemy that is closest to how databases will work, is that right?

3. I was initially thinking on the possibility of adding "standard" (sync) SQLAlchemy support to DatabaseMiddleware, or an equivalent SyncDatabaseMiddleware, to be able to use these SQLAlchemy models, and have them integrated directly into Starlette (and inherit that in FastAPI), knowing that they would only work properly on non-async functions. But as I'm understanding, right now that would not align with the current plan of having pure databases handled directly in code, right?

[tomchristie]

You might find this helpful: https://github.com/encode/starlette/pull/390

1. Yes. I've already done that in the docs now: https://www.starlette.io/database/
2. No. Individual connections lifetimes are handled for you automatically. (Short version: They'll autocommit as soon as possible unless they're inside a transaction.)

You will want to connect/disconnect the database connection pool on startup/shutdown.

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

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

1. It'd be like this...

@app.route("/notes", methods=["GET"])
async def list_notes(request):
    query = notes.select()
    results = await database.fetch_all(query)
    return JSONResponse(content)

1. Nope, you don't call .connect() and .disconnect(). Connections are managed for you, and you can wrap up operations in a transaction if you want to isolate a set of operations. Contection/Transaction management is dealt with using task-local storage, so databases is always able to find out whatever the current connection is.

2. "On the other side, I want to have docs of how to use "standard" SQLAlchemy ORM models in FastAPI" - You can't use SQLAlchemy ORM with any async web framework. Or at least, you can, but you need to run the operations within a threadpool. If you're using Starlette and you just write an endpoint as a standard function (non-async) then it'll do that for you and you can use SQLAlchemy from there just fine. You're then in a standard threaded context.

3. Yes exactly. I think you'd need to explicitly start and close an SQLAlchemy "session" once you're in the running-in-a-threadpool block.

4. An SQLAlchemy middleware would be really good, and would be something we'd certainly be able to recommend in Starlette. Thinking about it a bit more, the session setup/close might have to run inside the same thread as the endpoint, since I'm pretty sure SQLAlchemy ORM must use thread-local storage. If that's the case then you might need to start out implementing it as a decorator. (Perhaps have a look if there's anything like that already, that wraps a function up inside a single SQLAlchemy session) - It'd be really good to also be able to recommend a good integration for working with SQLAlchemy/sync-views, as an alternative to the low-level databases/async-views option. (Or indeed GINO or whatever else)

[tomchristie]

Closing this off now, in favor of http://github.com/encode/databases

[tomchristie]

Oh - one other caveat. I'm not sure you'd interact with SQLAlchemy inside middleware. Perhaps instead of relying on it's thread-local storage you're able to pass the session around explicitly instead or something - haven't looked into it sufficiently to be able to say.

[tiangolo]

Awesome, thanks for the thorough response.

---

Here's the extra info I've found regarding SQLAlchemy, how it uses threads, etc, as I understand it, in case it's useful (for future reference):

scoped_session

The "default" suggestion by SQLAlchemy is to use a scoped_session, it uses thread-local variables to make sure there's a single session per thread and the scoped_session is a proxy to a "registy" of Session objects, these Session objects are in this registry on a connection pool and are re-used and passed around by each thread, with a limit of a maximum number of Session objects. It works like that because it assumes a single request would be handled by a single thread, so, in that case, it would work. But the recommendation is to have a single session per request.

This is what I have in some of the examples in FastAPI but, although works for small examples, it's not working for bigger projects (I discovered it while creating a full project generator with PostgreSQL), it seems the sessions are not garbage collected so they are not being closed, and the limits are reached, etc.

explicit Session

But then, if I create the Session explicitly, by hand (in code) without using the "magic" scoped_session that uses thread-locals, I can pass that same object to the threadpool, and then I can close the session in the middleware (outside of the threadpool/different thread). The Session is not threadsafe, but as it is used on a single request, only on one thread at a time, I expect it not to have race conditions.

request state

But then, I need to come back to your previous idea of attaching the session to the request, as it might be on different threads even for the same request (for the middleware).

This is what is working (or seems to be working, as my confidence in understanding how to handle SQLAlchemy lowers), using your hack of request._scope:

from fastapi import FastAPI
from sqlalchemy import Boolean, Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from sqlalchemy.orm import sessionmaker
from starlette.requests import Request

# SQLAlchemy specific code, as with any other app
SQLALCHEMY_DATABASE_URI = "sqlite:///./test.db"
# SQLALCHEMY_DATABASE_URI = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URI, connect_args={"check_same_thread": False}
)
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)


class CustomBase:
    # Generate __tablename__ automatically
    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()


Base = declarative_base(cls=CustomBase)


class User(Base):
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean(), default=True)


Base.metadata.create_all(bind=engine)

db_session = Session()

first_user = db_session.query(User).first()
if not first_user:
    u = User(email="[email protected]", hashed_password="notreallyhashed")
    db_session.add(u)
    db_session.commit()

db_session.close()


# Utility
def get_user(db_session, user_id: int):
    return db_session.query(User).filter(User.id == user_id).first()


# FastAPI specific code
app = FastAPI()


@app.get("/users/{user_id}")
def read_user(request: Request, user_id: int):
    user = get_user(request._scope["db"], user_id=user_id)
    return user


@app.middleware("http")
async def close_db(request, call_next):
    request._scope["db"] = Session()
    response = await call_next(request)
    request._scope["db"].close()
    return response

[tomchristie]

Right. What * might* be neat would be if you could use explicit sessions in middleware, and then set the session scope once you enter the endpoint. Or perhaps figure how to get task-local sessions that are still task-local aware when a thread pool execution is spawned.

[tomchristie]

Anyways, good stuff there. I expect we’ll support a request.state interface soon, which would work well for your use case.

[tiangolo]

Awesome, thanks.

I hope to add a PR with a simple example using SQLAlchemy models to the docs here, after request.state is there... unless I find a way to make task-local sessions work across threads before that (if that seems useful here).

[tomchristie]

Probably let’s not over complicate things. request.state.db is good enough to get started with.

[sandys]

hi guys,
just my $0.02 .

I come from a standpoint of considerable pain around configuring sqlalchemy within flask. Would like to point some things out. Its not just about how to use databases within starlette. Its also related to how I can use other things as well.

1. the starlette/sqlalchemy models that i have written, should be usable from RQ/Celery. Is the code so intermingled that celery is having to instantiate a starlette instance ? This is a frequent mess in flask - a lot of sqlalchemy is done within "requestcontext" and causes a lot of problems when i load the models in celery/rq. Which is why you use "has_request_context" above. I can use the same code to load models into rq/celery.

2. connection pooling - mysql guys generally use the built in pooler. postgresql guys generally use pgbouncer. do note the problem is not around configuration in sqlalchemy itself. if you dont setup the configuration of a pooler the correct way (by respecting the threads, workers, etc) you again run into tons of "connection was closed problem".

I was struggling to find the answers to some of these questions.

[tomchristie]

Now that we've got a place for it, I'll start promoting https://discuss.encode.io/c/starlette as a good place for open-ended discussion and usage questions, but let's answer this here for now.

1. No. There's a very strict separation of concerns. (Eg. databases being a completely independent package.)
2. databases deals with connection pooling for you.

[sandys]

Hi Tom,
Should I continue this discussion here or on the other forum ? Don't want
to intrude in the wrong place.

Because I kinda disagree on the replacement of something like pgbouncer for
the application level pooling here.

Do note that larger teams have myriad sets of applications - so it's really
an infrastructure wide pooler.

On Sat, 22 Jun, 2019, 01:45 Tom Christie, [e-mail] wrote:

Now that we've got a place for it, I'll start promoting
https://discuss.encode.io/c/starlette as a good place for open-ended
discussion and usage questions, but let's answer this here for now.

1. No. There's a very strict separation of concerns. (Eg. databases
   being a completely independent package.)
2. databases deals with connection pooling for you.

—
You are receiving this because you commented.
Reply to this email directly, view it on GitHub
https://github.com/encode/starlette/issues/370?email_source=notifications&email_token=AAASYUZ4ET7FKEQRR6LBSVLP3UZH3A5CNFSM4GTJZPV2YY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGODYJPNNA#issuecomment-504559284,
or mute the thread
https://github.com/notifications/unsubscribe-auth/AAASYUY6MA2DS4LZY2LHK2DP3UZH3ANCNFSM4GTJZPVQ
.

[tomchristie]

You could connect to pgbouncer if you wanted. I guess you’d probably want to bump the application side connection pool settings quite high then, and let pgbouncer handle the actual limiting connections onto the database.

[reinoldus]

I think this discussion led into creating these docs: https://fastapi.tiangolo.com/tutorial/sql-databases/#create-the-sqlalchemy-engine

My main issue is that I am using starlette with graphene and so my root query function returns and passes the "SQLAlchemyORMStuff" class down, which is an issue since I have to make use of ORM lazyload in the lower Graphene-Objects, but the session is already closed.

    @staticmethod
    def resolve_recommendations(parent, info: ResolveInfo):
        db = SessionLocal()
        ra = SQLAlchemyORMStuff(5, db_local=db)
        db.close()
        return ra

The obvious solution would be to pre-calculate everything, but that kind of takes the beauty of graphene. Alternatively I could close the session in the destructor of "SQLAlchemyORMStuff", but that kind of leaves it to the garbage collector.

Anyone having an idea on how to tackle this? Might it be safe to not close the session and set a short timeout? I am a bit cautious here, because I already had sqlalchemy make my entire server unresponsive which is not fun.

Thank you for your help!

[succerror]

Probably let’s not over complicate things. request.state.db is good enough to get started with.

Now that we have request.state(Thanks for implementing it btw)

I have this simple middleware and It seems to work fine

class DBSessionMiddleware(BaseHTTPMiddleware):
    """Creates a database session and attaches it to
    `request.state.db` and auto commits and closes it at
    the end of the response.
    If there is an exception in processing the request,
    session will be rolled back unless explicitly committed
    inside the view.
    """
    async def dispatch(self, request, call_next):
        request.state.db = Session()
        try:
            response = await call_next(request)
            request.state.db.commit()
        except Exception:
            request.state.db.rollback()
            raise
        finally:
            request.state.db.close()
        return response

Then the view be like

async def add_book(request):
    db = request.state.db
    data = await request.json()
    book = Book(name=data['name'], type=data['type'])
    db.add(book)
    return Response(status_code=200)

[jpic]

Thank you so much for the discussion, also to @succerror for the example that got me started. I have a couple of tricks to share:

• do this in an ASGI Middleware and you also get WebSocket endpoint support
• use contextvars to magically abstract the SQLA session away and use models more like Django

Here is an example, using SQLA 1.4b1 which supports async:

import contextvars    
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine    

engine = create_async_engine('postgresql://localhost/starmvc', echo=True)
async_session = contextvars.ContextVar('async_session')    

starlette = Starlette(**yourstuff)

async def app(scope, receive, send):                                                           
    async with AsyncSession(engine) as session:                                                
        token = async_session.set(session)                                                     
        try:                                                                                   
            async with session.begin():                                                        
                response = await starlette(scope, receive, send)                               
            await session.commit()                                                             
        except:                                                                                
            await session.rollback()                                                           
            raise                                                                              
        finally:                                                                               
            await session.close()                                                              
            async_session.reset(token)                                                         
        return response     

Note that you don't need the async_session contextvars, it's only useful if you want to abstract away the SQLA session object like so:

class AsyncSessionDescriptor:    
    def __get__(self, obj, objtype):      
        session = async_session.get(False)                                                     
        assert session, 'starmvc.orm.async_session was not set'
        # TODO: do something better than the above line ^^
        return session                                                                         


class QueryDescriptor:                                                                         
    def __get__(self, obj, objtype):                                                           
        return objtype._session.sync_session.query(objtype)                                    


class ModelBase:                                                                               
    _session = AsyncSessionDescriptor()                                                        
    objects = QueryDescriptor()                                                                

    @classmethod                                                                               
    def create(cls, **kwargs):                                                                 
        obj = cls(**kwargs)                                                                    
        obj.save()                                                                             
        return obj                                                                             

    def delete(self):                                                                          
        self._session.delete(self)

    def save(self):
        self._session.add(self)


Model = declarative_base(cls=ModelBase)

I'm sorry for not presenting code that is as polished as yours, but I hope this can still be useful to others who are trying to get started.

This has not been tested in production. However, here is an additional trick you can take for free: pytest support

    @pytest.fixture(autouse=True)
    async def orm_async_session():
        async with AsyncSession(engine) as session:
            async_session.set(session)
            yield session
            await session.rollback()

    @pytest.fixture(scope='session', autouse=True)    
    async def migrate():    
        async with create_your_engine().begin() as conn:    
            await conn.run_sync(YourModel.metadata.create_all)    
        yield    
        async with engine.begin() as conn:    
            await conn.run_sync(YourModel.metadata.drop_all)    

    @pytest.fixture(scope='session')
    def event_loop(request):
        """Fix for the following error:

        ScopeMismatch: You tried to access the 'function' scoped fixture
        'event_loop' with a 'session' scoped request object, involved factories
        """
        loop = asyncio.get_event_loop_policy().new_event_loop()
        yield loop
        loop.close()


    # monkey patch the event loop yay
    from pytest_asyncio import plugin
    plugin.event_loop = event_loop

However, this will require a patched version of pytest-asyncio which maintains the contextvars for a test. But this will allow the User model to work outside of the requests, which allows you to test like with Django:

@pytest.mark.asyncio
async def test_user_create(client):
    response = await client.post(
        '/user/create',
        data=dict(
            email='[email protected]',
        ),
    )
    assert response.status_code == 200
    assert User.objects.where(User.email == '[email protected]').count()