升级SQLAlchemy到2.0.19

models/bilibili.py

@@ -1,11 +1,14 @@
 # -*- coding: utf-8 -*-
+import datetime
+
 import sqlalchemy
+from sqlalchemy.orm import Mapped, mapped_column
 
 import models.database
 
 
 class BilibiliUser(models.database.OrmBase):
     __tablename__ = 'bilibili_users'
-    uid = sqlalchemy.Column(sqlalchemy.BigInteger, primary_key=True)  # 创建表后最好手动改成unsigned
+    uid: Mapped[int] = mapped_column(sqlalchemy.BigInteger, primary_key=True)  # 创建表后最好手动改成unsigned
-    avatar_url = sqlalchemy.Column(sqlalchemy.String(100))
+    avatar_url: Mapped[str] = mapped_column(sqlalchemy.String(100))
-    update_time = sqlalchemy.Column(sqlalchemy.DateTime)
+    update_time: Mapped[datetime.datetime]

models/database.py

@@ -1,25 +1,25 @@
 # -*- coding: utf-8 -*-
 from typing import *
 
-import sqlalchemy.ext.declarative
 import sqlalchemy.orm
 
 import config
 
-OrmBase = sqlalchemy.ext.declarative.declarative_base()
+_engine: Optional[sqlalchemy.Engine] = None
-_engine = None
+
-_DbSession: Optional[Type[sqlalchemy.orm.Session]] = None
+
+class OrmBase(sqlalchemy.orm.DeclarativeBase):
+    pass
 
 
 def init(_debug):
     cfg = config.get_config()
-    global _engine, _DbSession
+    global _engine
     # engine = sqlalchemy.create_engine(cfg.database_url, echo=debug)
     _engine = sqlalchemy.create_engine(cfg.database_url)
-    _DbSession = sqlalchemy.orm.sessionmaker(bind=_engine)
 
     OrmBase.metadata.create_all(_engine)
 
 
-def get_session():
+def get_session() -> sqlalchemy.orm.Session:
-    return _DbSession()
+    return sqlalchemy.orm.Session(_engine)

requirements.txt

@@ -1,4 +1,4 @@
 -r blivedm/requirements.txt
 pycryptodome==3.10.1
-sqlalchemy==1.4.31
+sqlalchemy==2.0.19
 tornado==6.3.2

services/avatar.py

@@ -8,7 +8,6 @@ import urllib.parse
 from typing import *
 
 import aiohttp
-import sqlalchemy
 import sqlalchemy.exc
 
 import config
@@ -80,8 +79,10 @@ def get_avatar_url_from_database(user_id) -> Awaitable[Optional[str]]:
 def _do_get_avatar_url_from_database(user_id):
     try:
         with models.database.get_session() as session:
-            user = session.query(bl_models.BilibiliUser).filter(
+            user = session.scalars(
+                sqlalchemy.select(bl_models.BilibiliUser).filter(
                     bl_models.BilibiliUser.uid == user_id
+                )
             ).one_or_none()
             if user is None:
                 return None
@@ -285,8 +286,10 @@ def _update_avatar_cache_in_memory(user_id, avatar_url):
 def _update_avatar_cache_in_database(user_id, avatar_url):
     try:
         with models.database.get_session() as session:
-            user = session.query(bl_models.BilibiliUser).filter(
+            user = session.scalars(
+                sqlalchemy.select(bl_models.BilibiliUser).filter(
                     bl_models.BilibiliUser.uid == user_id
+                )
             ).one_or_none()
             if user is None:
                 user = bl_models.BilibiliUser(