Upgrade to SQLA v2 (#1250)

Author: lmmx

.github/workflows/ci.yml

@@ -75,7 +75,7 @@ jobs:
         make install
     - name: Run tests
       run: |
-        make test
+        PYTHONASYNCIODEBUG=1 make test
 
   doc:
     name: Documentation

demos/blog/aiohttpdemo_blog/db.py

@@ -1,87 +1,83 @@
-from datetime import datetime as dt
-
-import asyncpgsa
-from sqlalchemy import (
-    MetaData, Table, Column, ForeignKey,
-    Integer, String, DateTime
+from datetime import datetime
+
+from sqlalchemy import ForeignKey, String
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from sqlalchemy.orm import (
+    DeclarativeBase,
+    Mapped,
+    mapped_column,
+    relationship,
+    selectinload,
 )
 from sqlalchemy.sql import select
 
-metadata = MetaData()
 
+class Base(DeclarativeBase):
+    pass
 
-users = Table(
-    'users', metadata,
 
-    Column('id', Integer, primary_key=True),
-    Column('username', String(64), nullable=False, unique=True),
-    Column('email', String(120)),
-    Column('password_hash', String(128), nullable=False)
-)
+class Users(Base):
+    __tablename__ = "users"
 
+    id: Mapped[int] = mapped_column(primary_key=True)
+    username: Mapped[str] = mapped_column(String(64), nullable=False, unique=True)
+    email: Mapped[str] = mapped_column(String(120))
+    password_hash: Mapped[str] = mapped_column(String(128), nullable=False)
 
-posts = Table(
-    'posts', metadata,
+    posts: Mapped[list["Posts"]] = relationship(
+        back_populates="user", lazy="raise_on_sql"
+    )
 
-    Column('id', Integer, primary_key=True),
-    Column('body', String(140)),
-    Column('timestamp', DateTime, index=True, default=dt.utcnow),
 
-    Column('user_id',
-           Integer,
-           ForeignKey('users.id'))
-)
+class Posts(Base):
+    __tablename__ = "posts"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    body: Mapped[str] = mapped_column(String(140))
+    timestamp: Mapped[datetime] = mapped_column(index=True, default=datetime.utcnow)
+
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
+    user: Mapped[Users] = relationship(back_populates="posts", lazy="raise_on_sql")
 
 
 async def init_db(app):
-    dsn = construct_db_url(app['config']['database'])
-    pool = await asyncpgsa.create_pool(dsn=dsn)
-    app['db_pool'] = pool
-    return pool
+    dsn = construct_db_url(app["config"]["database"])
+    engine = create_async_engine(dsn)
+    app["db_pool"] = async_sessionmaker(engine)
+
+    yield
+
+    await engine.dispose()
 
 
 def construct_db_url(config):
-    DSN = "postgresql://{user}:{password}@{host}:{port}/{database}"
+    DSN = "postgresql+asyncpg://{user}:{password}@{host}:{port}/{database}"
     return DSN.format(
-        user=config['DB_USER'],
-        password=config['DB_PASS'],
-        database=config['DB_NAME'],
-        host=config['DB_HOST'],
-        port=config['DB_PORT'],
+        user=config["DB_USER"],
+        password=config["DB_PASS"],
+        database=config["DB_NAME"],
+        host=config["DB_HOST"],
+        port=config["DB_PORT"],
     )
 
 
-async def get_user_by_name(conn, username):
-    result = await conn.fetchrow(
-        users
-        .select()
-        .where(users.c.username == username)
-    )
+async def get_user_by_name(sess, username):
+    result = await sess.scalar(select(Users).where(Users.username == username))
     return result
 
 
-async def get_users(conn):
-    records = await conn.fetch(
-        users.select().order_by(users.c.id)
-    )
-    return records
+async def get_users(sess):
+    records = await sess.scalars(select(Users).order_by(Users.id))
+    return records.all()
 
 
-async def get_posts(conn):
-    records = await conn.fetch(
-        posts.select().order_by(posts.c.id)
-    )
-    return records
+async def get_posts(sess):
+    records = await sess.scalars(select(Posts).order_by(Posts.id))
+    return records.all()
 
 
-async def get_posts_with_joined_users(conn):
-    j = posts.join(users, posts.c.user_id == users.c.id)
-    stmt = select(
-        [posts, users.c.username]).select_from(j).order_by(posts.c.timestamp)
-    records = await conn.fetch(stmt)
-    return records
-
-
-async def create_post(conn, post_body, user_id):
-    stmt = posts.insert().values(body=post_body, user_id=user_id)
-    await conn.execute(stmt)
+async def get_posts_with_joined_users(sess):
+    records = await sess.scalars(
+        select(Posts).options(selectinload(Posts.user)).order_by(Posts.timestamp)
+    )
+    return records.all()

demos/blog/aiohttpdemo_blog/db_auth.py

@@ -4,16 +4,14 @@
 
 
 class DBAuthorizationPolicy(AbstractAuthorizationPolicy):
-
-    def __init__(self, db_pool):
-        self.db_pool = db_pool
+    def __init__(self, app):
+        self.app = app
 
     async def authorized_userid(self, identity):
-        async with self.db_pool.acquire() as conn:
-            user = await db.get_user_by_name(conn, identity)
+        async with self.app["db_pool"]() as sess:
+            user = await db.get_user_by_name(sess, identity)
             if user:
                 return identity
-
         return None
 
     async def permits(self, identity, permission, context=None):

demos/blog/aiohttpdemo_blog/forms.py

@@ -16,7 +16,7 @@ async def validate_login_form(conn, form):
 
     if not user:
         return 'Invalid username'
-    if not check_password_hash(password, user['password_hash']):
+    if not check_password_hash(password, user.password_hash):
         return 'Invalid password'
 
     return None

demos/blog/aiohttpdemo_blog/main.py

@@ -42,9 +42,10 @@ async def init_app(config):
 
     setup_routes(app)
 
-    db_pool = await init_db(app)
+    app.cleanup_ctx.append(init_db)
 
     redis = await setup_redis(app)
+    app.on_shutdown.append(lambda _: redis.aclose())
     setup_session(app, RedisStorage(redis))
 
     # needs to be after session setup because of `current_user_ctx_processor`
@@ -57,7 +58,7 @@ async def init_app(config):
     setup_security(
         app,
         SessionIdentityPolicy(),
-        DBAuthorizationPolicy(db_pool)
+        DBAuthorizationPolicy(app)
     )
 
     log.debug(app['config'])

demos/blog/aiohttpdemo_blog/templates/index.html

@@ -3,6 +3,6 @@
 {% block content %}
     <h1>Hi, {{ user.username }}!</h1>
     {% for post in posts %}
-    <div><p>[{{ post.timestamp.strftime('%Y-%m-%d %H:%M:%S') }}] {{ post.username }} posted: <b>{{ post.body }}</b></p></div>
+    <div><p>[{{ post.timestamp.strftime('%Y-%m-%d %H:%M:%S') }}] {{ post.user.username }} posted: <b>{{ post.body }}</b></p></div>
     {% endfor %}
 {% endblock %}

demos/blog/aiohttpdemo_blog/views.py

@@ -17,9 +17,9 @@ async def index(request):
     if not username:
         raise redirect(request.app.router, 'login')
 
-    async with request.app['db_pool'].acquire() as conn:
-        current_user = await db.get_user_by_name(conn, username)
-        posts = await db.get_posts_with_joined_users(conn)
+    async with request.app['db_pool']() as sess:
+        current_user = await db.get_user_by_name(sess, username)
+        posts = await db.get_posts_with_joined_users(sess)
 
     return {'user': current_user, 'posts': posts}
 
@@ -33,16 +33,16 @@ async def login(request):
     if request.method == 'POST':
         form = await request.post()
 
-        async with request.app['db_pool'].acquire() as conn:
-            error = await validate_login_form(conn, form)
+        async with request.app['db_pool']() as sess:
+            error = await validate_login_form(sess, form)
 
             if error:
                 return {'error': error}
             else:
                 response = redirect(request.app.router, 'index')
 
-                user = await db.get_user_by_name(conn, form['username'])
-                await remember(request, response, user['username'])
+                user = await db.get_user_by_name(sess, form['username'])
+                await remember(request, response, user.username)
 
                 raise response
 
@@ -64,9 +64,9 @@ async def create_post(request):
     if request.method == 'POST':
         form = await request.post()
 
-        async with request.app['db_pool'].acquire() as conn:
-            current_user = await db.get_user_by_name(conn, username)
-            await db.create_post(conn, form['body'], current_user['id'])
+        async with request.app['db_pool'].begin() as sess:
+            current_user = await db.get_user_by_name(sess, username)
+            sess.add(db.Posts(body=form["body"], user_id=current_user.id))
             raise redirect(request.app.router, 'index')
 
     return {}