How to keep single session with reopening connections?

[albertalexandrov]

Hi there!

P.S. Please forgive me if my wording seems strange.

Please look at the code:

import asyncio
from contextvars import ContextVar

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

engine = create_async_engine("postgresql+asyncpg://postgres:postgres@localhost:5433/postgres", isolation_level="AUTOCOMMIT")
session_context = ContextVar("session")


class Repository:

    async def get_one(self):
        statement = text("select 1")
        return await self.execute(statement)

    async def get_two(self):
        statement = text("select 2")
        return await self.execute(statement)

    async def execute(self, statement):
        async with AsyncSession(engine) as session:
            print(session)
            return (await session.execute(statement)).scalar()


async def main():
    repo = Repository()
    print(await repo.get_one())
    print(await repo.get_two())


if __name__ == '__main__':
    asyncio.run(main())

It reproduces the output:

<sqlalchemy.ext.asyncio.session.AsyncSession object at 0x113e3d3d0>
1
<sqlalchemy.ext.asyncio.session.AsyncSession object at 0x113b43380>
2

As you could see sessions at each execute method call are different.

How could I make session be the same but at the same time close connection each execute method call (to release the connection)?

I mean calling get_one method, create session first time, make query, close connection. Then call method get_two, reuse session created during get_one method, somehow reopen connection, make query, close connection.

As you could guess I need a single session objects during some time (f.e. web request).

Thanks in advance and happy new year!

[zzzeek]

the ORM Session used to support this directly through a feature called "autocommit" mode, we later named this to "library level autocommit mode" to clarify its difference from the driver level autocommit that you're using above. This mode was deprecated in the 1.4 series of SQLAlchemy and removed in 2.0. It basically led to code that created lots of connections and these connections were continuously having rollback/commit called on them, which performs poorly and also led to more deadlocks and transactional consistency issues.

there might be ways to rig a Session with events to sort of act in this way again but if you are using driver level autocommit with PostgreSQL, the connections wont be "idle in transaction" or anything like that and there's no real reason, especially in a single web request, to use more than one connection. just use the one connection. if this is some kind of long-running request, where by "long running" I mean tens of seconds to minutes or more, you would want to be closing out the session for those long pauses as you already are doing. You can also call Session.commit() which, if you have expire_on_commit=False which I think you definitely should here, that will close out connectivity resources used by the session without removing the actual objects you have associated with it. We dont yet have a method on session that simply drops all connectivity without either committing present state or otherwise expiring/expunging all the objects within, though a parameter on session.reset() would likely be able to do this

[albertalexandrov]

Oh, yes, @zzzeek , I did forgot about expire_on_commit. I use it in every project. My bad. Thanks!

Could you be so kind to answer another question. I modified the code above a bit (it also on StackOverflow).

from contextvars import ContextVar

from fastapi import FastAPI, BackgroundTasks
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

engine = create_async_engine("postgresql+asyncpg://postgres:postgres@localhost:5433/postgres", isolation_level="AUTOCOMMIT")
session_context = ContextVar("session")

app = FastAPI()


class Repository:

    async def get_one(self):
        return await self.execute(text("select 1"))

    async def get_two(self):
        return await self.execute(text("select 2"))

    async def execute(self, statement):
        try:
            session = session_context.get()
        except LookupError:
            session = AsyncSession(engine)
            session_context.set(session)
        print(session)
        result = (await session.execute(statement)).scalar()
        await session.close()  # for some reason I need to close session every time
        return result


async def check_connections_statuses():
    print(engine.pool.status())
    print(session_context.get())


@app.get("/")
async def main(background_tasks: BackgroundTasks):
    repo = Repository()
    print(await repo.get_one())
    print(await repo.get_two())
    background_tasks.add_task(check_connections_statuses)

The code produces output:

<sqlalchemy.ext.asyncio.session.AsyncSession object at 0x121bb31a0>
1
<sqlalchemy.ext.asyncio.session.AsyncSession object at 0x121bb31a0>
2
INFO:     127.0.0.1:59836 - "GET / HTTP/1.1" 200 OK
Pool size: 5  Connections in pool: 1 Current Overflow: -4 Current Checked out connections: 0
<sqlalchemy.ext.asyncio.session.AsyncSession object at 0x121bb31a0>

It seems that connections does not leak. But what about session object in ContextVar? As you could see it is still in ContextVar. Is it save not to reset it? I've searched for ContextVar using in this repository to see how you could use it and I've found only _orm_adapter_trace_context and it's not being resetted.

Thanks in advance!

[CaselIT]

if you close it every time like you are doing it should be fine to leave it in the context var.

Usually people using fastapi use a depencency to manage the session, avoiding the need of using a contextvar.
I'm no expect on fastapi, but I think it's something like what's documented here https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-with-yield/

[albertalexandrov]

@zzzeek , @CaselIT , thanks a lot :)