Sanic: Integration with SQLAlchemy

[teamhide]

I'm new at sanic and async paradigm in python.

In general python project, that used mysql, usually used SQLAlchemy for ORM.

I wonder that should I using pure SQLAlchemy to ORM for sanic?

Or, I have to use GINO?

Any comments, appreciate :)

[ahopkins]

You can use either. The benefit of an async ORM (like GINO) is that it leverages the power of async enabled applications. Meaning that when a request comes in that requires a trip to the DB, that worker is still free to accept incoming requests before the DB query returns. It is kit necessarily that your script runs faster, but more efficiently.

With that said, you need to decide whether the tradeoff is worth it for you. A package like SQLAlchemy has far more resources available for you.

I'm closing for now, but feel free to keep the conversation going.

[teamhide]

@ahopkins Can you provide some documents related about this or example?

[chenjr0719]

@teamhide This might help.
https://github.com/huge-success/sanic/wiki/Examples#aiopeewee

[vltr]

@teamhide the biggest change for me in using SQLAlchemy with an async framework like Sanic is how to properly deal with the Session during the request lifecycle. You can read more in the SQLAlchemy documentation. Basically, what I did in my projects was to create a Session and append it to the current request object, example:

@app.url("/")
async def my_handler(request):
    request["session"] = Session()  # request has a dict like interface
    ....

Then, I could use request["session"] everytime I wanted inside the request lifecycle. To dispose the Session afterwards, it's just a matter of creating a response middleware to close any session (if exists):

@app.middleware("response")
async def close_session(request, response):
    if "session" in request:
        try:
            request["session"].commit()
        finally:
            request["session"].remove()

Note that, in the above example, you can add some more fined grained error control if something went wrong :wink:

I hope this helps!

[teamhide]

Thanks for detailed comments everyone :) 👍👍👍

[Senhaji-Rhazi-Hamza]

@teamhide the biggest change for me in using SQLAlchemy with an async framework like Sanic is how to properly deal with the Session during the request lifecycle. You can read more in the SQLAlchemy documentation. Basically, what I did in my projects was to create a Session and append it to the current request object, example:

@app.url("/")
async def my_handler(request):
    request["session"] = Session()  # request has a dict like interface
    ....

Then, I could use request["session"] everytime I wanted inside the request lifecycle. To dispose the Session afterwards, it's just a matter of creating a response middleware to close any session (if exists):

@app.middleware("response")
async def close_session(request, response):
    if "session" in request:
        try:
            request["session"].commit()
        finally:
            request["session"].remove()

Note that, in the above example, you can add some more fined grained error control if something went wrong

I hope this helps!

The problem with this example, is that I'm not sure if we have guaranties that one request is associated to one thread in asyncio, I think that multiple asyncio requests, can run in the same thread, I'm not sure what the result will be ? but thank you anyway for the example

[ahopkins]

Sanic workers run entirely single threaded. Unless you are pushing something off to a thread yourself, Sanic will not do that for you.

As an FYI, that example is a little outdated now. You should use:

request.ctx.session

[Senhaji-Rhazi-Hamza]

Sanic workers run entirely single threaded. Unless you are pushing something off to a thread yourself, Sanic will not do that for you.

As an FYI, that example is a little outdated now. You should use:

request.ctx.session

First of all, you are doing great work with Sanic, it's simple, not much opinionated as other "micro" frameworks (also liked the podcast talks you have given about it)

So I'm understanding that I don't need to use the scope_func argument here : https://docs.sqlalchemy.org/en/13/orm/contextual.html#using-custom-created-scopes

I think what would be ideal (although I'm not sure), would be to have a function, that returns an id of the current request processed like here with app_ctx_stact._ident_func_ in : https://towardsdatascience.com/use-flask-and-sqlalchemy-not-flask-sqlalchemy-5a64fafe22a4 for flask

I have seen in the latest documentation something that could serve this purpose : https://sanic.readthedocs.io/en/latest/sanic/api_reference.html#sanic.request.Request.id

But i think is not released yet

The benefit would be, that by having an id or a function that returns the id of the current request, it would decouple the problem of the necessity of having the guaranty that the request is treated by one thread, since the session would be now associated to the request and not the current thread anymore

Lets me know if it makes any sens, thank you !

[ahopkins]

Thank you very much, I really appreciate that.

Version 21.3 is set for release in a couple weeks with request.id as you mentioned. If you need that functionality now, you have a few options that come to mind.

1. Install from master
2. Use id(request)
3. Create a custom Request class
4. Add it in middleware

Number 3 would look something like this:

from sanic.request import Request
import uuid

class CustomRequest(Request):
    @property
    def id(self):
        return uuid.uuid4()

app = Sanic("myapp", request_class=CustomRequest)

Number 4 something like this:

@app.middleware
def add_request_id(request):
    request.ctx.id = uuid.uuid4()

[Senhaji-Rhazi-Hamza]

Thank you very much, I really appreciate that.

Version 21.3 is set for release in a couple weeks with request.id as you mentioned. If you need that functionality now, you have a few options that come to mind.

1. Install from master
2. Use id(request)
3. Create a custom Request class
4. Add it in middleware

Number 3 would look something like this:

from sanic.request import Request
import uuid

class CustomRequest(Request):
    @property
    def id(self):
        return uuid.uuid4()

app = Sanic("myapp", request_class=CustomRequest)

Number 4 something like this:

@app.middleware
def add_request_id(request):
    request.ctx.id = uuid.uuid4()

Thank you a lot for the quality of feedback, the proposed solutions, solve my constraint

[vltr]

@Senhaji-Rhazi-Hamza I think I wrote the first answer to this question a while ago. Even though I'm not using SQLAlchemy all that often, I'm shifting my development to use it again. But, SQLAlchemy is not asynchronous (as you already know) and indeed using scope_func argument with the scoped_session call is what I used in the past with Sanic and even Twisted Matrix - back when I was still writing Python 2 code.

The simplest solution I have is to basically use the id function from the Python locals to identify a request (if that's where your session lifespan will exist), like:

def _get_ident(request):
    return id(request)


async def something(request):
    request.ctx.session_factory = scoped_session(sessionmaker(...), scopefunc=lambda: _get_ident(request))
    session = request.ctx.session_factory()
    ...

then, on a response middleware, I can try to close any sessions that were possibily created, just like the example I made:

@app.middleware("response")
async def close_session(request, response):
    if request.ctx.session_factory:
        try:
            request.ctx.session_factory.remove()
        finally:  # or catch Exception
            ...

one interesting idea would be to create a decorator for your async functions to inject the session already and close it after it has been called:

from functools import wraps


def inject_session():
    def outer(fn):
        @wraps(fn)
        async def inner(request, *args, **kwargs):
            session_factory = scoped_session(
                sessionmaker(...),
                scopefunc=lambda: _get_ident(request)  # I don't know if that's even required now
            )
            session = session_factory()
            kwargs.update({
                "session": session
            })
            response = await fn(request, *args, **kwargs)
            session.close()
            session_factory.remove()  # I don't know if that's actually necessary
            return response
        return inner
    return outer


@app.get("/")
@inject_session
async def my_index(request, session, *args, **kwargs):
    # use session as normal in here
    ...

you could wrap the endpoint call with try / except to add some control over transactions, do rollbacks, perform commits and so on ... I mean, there are a lot of possibilities, you just got to choose what's best for you, but, keep one thing in mind: state management in SQLAlchemy can be tricky if inside asynchronous apps, so a lot of testing and proper debugging to check if your chosen approach will not block other requests or leave loose ends that might not be collected by the gc and make your memory consumption increase with time ... but it's not an impossible task, be sure of that :wink:

NOTE: I did not tested the codes I wrote above, they might not work correctly or out of the box.

[Senhaji-Rhazi-Hamza]

@vltr That's really nice and genuine from you to provide those examples, I like the ideas behind.

That said, I'm more a fan of the approach where you can access the session directly from the request context, and you don't need to handle it as an argument in your view function, I'm also fan of the idea, that my model, can access the current session depending on the context at any time, and it's achievable by injecting the current session to the base model during the request handling for example :
python @app.middleware("request") async def inject_session(request): request.ctx.session_factory = scoped_session(sessionmaker(...), scopefunc=lambda: id(request)) request.ctx.session = request.ctx.session_factory() BaseModel.has_request_context=True BaseModel.request_ctx_session = request.ctx.session ```` python
class class_property(object):

# This class property is inspired from the behaviour of flaskSqlAlchemy
# The behaviour of querying directly a model User.query.. is nice, but
# the flaskSqlAlchemy dependency is not, so the behaviour is reproduced
# with this trick of class property
def __init__(self, fget):
    self.fget = fget

def __get__(self, owner_self, owner_cls):
    return self.fget(owner_cls)

class BaseModel(db.Base):

__abstract__ = True
creation_timestamp = db.Column(db.DateTime())

def __init__(self, **kwargs):
    super(BaseModel, self).__init__(**kwargs)
    self.creation_timestamp = datetime.utcnow()

@class_property
def query(cls):
    return cls.session.query(cls)

@class_property
def session(cls):
    if hasattr(cls, 'has_request_context') and cls.has_request_context:
        return cls.request_ctx_session
    return db.session

def save(self):
    self.session.add(self)
    self.session.commit()

````

with this approach the model handles automatically the session whether it is in a context request or outside

```python

async def something(request):
user_id = get_user_id_from_request(request)
user = User.get(user_id)
user.do_something() # here the BaseModel will choose the session depending on the context
````
NOTE : I haven't tested this too, but I have previously make similar logic work in flask

Thank you ! :D

[vltr]

@Senhaji-Rhazi-Hamza I'm glad this solution looks better for you! I'm not a big fan of the "Active Record" approach, but we're here to help on issues :sunglasses: But, I do like that exactly class_property snippet, I think I might even have it somewhere on some of my old codes!

Anyway, one thing I noticed: during the lifespan of a request and setting the session directly to the BaseModel class might give you some issues because "concurrently", within the asyncio current task, it might close one session from one request where another concurrent request might be still using it (I know you did not closed the session as far as I saw in your examples, but most of the times you might want to do that).

To avoid this issue, I would suggest you to use contextvars instead, where a session would be bound to the current asyncio task instead of the one in your class. You can use contextvars on Python 3.7+ or use aiocontextvars for previous versions of Python (such as 3.5 or 3.6):

from contextvars import ContextVar


_base_model_session_ctx = ContextVar("base_model_session_ctx")


class class_property(object):

    # This class property is inspired from the behaviour of flaskSqlAlchemy
    # The behaviour of querying directly a model User.query.. is nice, but
    # the flaskSqlAlchemy dependency is not, so the behaviour is reproduced
    # with this trick of class property
    def __init__(self, fget):
        self.fget = fget

    def __get__(self, owner_self, owner_cls):
        return self.fget(owner_cls)


class BaseModel(db.Base):

    __abstract__ = True
    creation_timestamp = db.Column(db.DateTime())

    def __init__(self, **kwargs):
        super(BaseModel, self).__init__(**kwargs)
        self.creation_timestamp = datetime.utcnow()

    @class_property
    def query(cls):
        return cls.session.query(cls)

    @class_property
    def session(cls):
        if _base_model_session_ctx.get(None) is not None:
            return _base_model_session_ctx.get()
        return db.session

    def save(self):
        self.session.add(self)
        self.session.commit()


@app.middleware("request")
async def inject_session(request):
    request.ctx.session_factory = scoped_session(sessionmaker(...), scopefunc=lambda: id(request))
    request.ctx.session = request.ctx.session_factory()
    request.ctx.session_ctx_token = _base_model_session_ctx.set(request.ctx.session)


@app.middleware("response")
async def close_session(request, response):
    if "session_ctx_token" in request.ctx:
        _base_model_session_ctx.reset(request.ctx.session_ctx_token)
        request.ctx.session.close()
        request.ctx.session_factory.remove()


async def something(request):
    user_id = get_user_id_from_request(request)
    user = User.get(user_id)
    user.do_something()

Again, not tested code, but you can now have some idea :wink:

[Senhaji-Rhazi-Hamza]

@Senhaji-Rhazi-Hamza I'm glad this solution looks better for you! I'm not a big fan of the "Active Record" approach, but we're here to help on issues But, I do like that exactly class_property snippet, I think I might even have it somewhere on some of my old codes!

Anyway, one thing I noticed: during the lifespan of a request and setting the session directly to the BaseModel class might give you some issues because "concurrently", within the asyncio current task, it might close one session from one request where another concurrent request might be still using it (I know you did not closed the session as far as I saw in your examples, but most of the times you might want to do that).

To avoid this issue, I would suggest you to use contextvars instead, where a session would be bound to the current asyncio task instead of the one in your class. You can use contextvars on Python 3.7+ or use aiocontextvars for previous versions of Python (such as 3.5 or 3.6):

from contextvars import ContextVar


_base_model_session_ctx = ContextVar("base_model_session_ctx")


class class_property(object):

    # This class property is inspired from the behaviour of flaskSqlAlchemy
    # The behaviour of querying directly a model User.query.. is nice, but
    # the flaskSqlAlchemy dependency is not, so the behaviour is reproduced
    # with this trick of class property
    def __init__(self, fget):
        self.fget = fget

    def __get__(self, owner_self, owner_cls):
        return self.fget(owner_cls)


class BaseModel(db.Base):

    __abstract__ = True
    creation_timestamp = db.Column(db.DateTime())

    def __init__(self, **kwargs):
        super(BaseModel, self).__init__(**kwargs)
        self.creation_timestamp = datetime.utcnow()

    @class_property
    def query(cls):
        return cls.session.query(cls)

    @class_property
    def session(cls):
        if _base_model_session_ctx.get(None) is not None:
            return _base_model_session_ctx.get()
        return db.session

    def save(self):
        self.session.add(self)
        self.session.commit()


@app.middleware("request")
async def inject_session(request):
    request.ctx.session_factory = scoped_session(sessionmaker(...), scopefunc=lambda: id(request))
    request.ctx.session = request.ctx.session_factory()
    request.ctx.session_ctx_token = _base_model_session_ctx.set(request.ctx.session)


@app.middleware("response")
async def close_session(request, response):
    if "session_ctx_token" in request.ctx:
        _base_model_session_ctx.reset(request.ctx.session_ctx_token)
        request.ctx.session.close()
        request.ctx.session_factory.remove()


async def something(request):
    user_id = get_user_id_from_request(request)
    user = User.get(user_id)
    user.do_something()

Again, not tested code, but you can now have some idea

Hey @vltr Thank you for your feedback (sorry for the delay of mine), thank you a lot for the example provided.

My understanding may be a bit limited, but I have a little question here, since the session with scoped_session(sessionmaker(...), scopefunc=lambda: id(request)) is bounded to the request, how is it possible that the problem you described here "within the asyncio current task, it might close one session from one request where another concurrent request might be still using it" might happen

In my understanding, no other request may close another session except its own ? (a relation one--one with session--request)

Thank you !

[vltr]

My understanding may be a bit limited, but I have a little question here, since the session with scoped_session(sessionmaker(...), scopefunc=lambda: id(request)) is bounded to the request, how is it possible that the problem you described here "within the asyncio current task, it might close one session from one request where another concurrent request might be still using it" might happen

now it's me asking for sorry :sweat_smile: so, the problem in here is not with the scopefunc: it's with the class_property session being used from a second or any other request while it was only created by the first one - which might lead to some problems because either the first request will remain in memory because the session is being used or might be closed and subsequent requests that thinks there's a session, will end up having an error.

it's kind of a mess to describe a fast paced timeline of events, but I can try and do better if you still have any problems on finding where you might have an issue with that approach. I mean, you might have, not that you will have, it's just my moto to be safe rather than sorry :sunglasses:

drop a line if you need some help! cheers!

[Senhaji-Rhazi-Hamza]

Hey @vltr ! Good explanation, it didn't cross my mind that I might have an issue with class_property, thanks a lot !! (y) (y) :D !

Cheers !