fastapi sqlalchemy 多线程使用

SingYi

发布于 2022-07-13 20:37:59

1.7K0

发布于 2022-07-13 20:37:59

文章被收录于专栏：Lan小站

由于业务需求，抓取数据后直接由python端入库，然后为了效率考虑多线程肯定是基本操作啦，之前为了提升抓取速度已经用上了异步，将速度由原来的20多s提到1s然后速度太快导致被滑块验证拦截，于是加上了代理IP。今天用我之前fastapi操作MySQL那套去操作数据库的时候，发现多线程并不好用，各种报错

经过百度，发现了一篇

sqlalchemy 多线程创建session:https://blog.csdn.net/Gragon_Shao/article/details/112786197

于是我把我之前用的那套模板改了一下

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session
from Config import config

if config.DBType == 'sqlite':
    # 使用SQLite数据库
    SQLALCHEMY_DATABASE_URL = f"sqlite:///{config.DataBase}"
    engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
    SessionLocal = sessionmaker(autocommit=False, autoflush=True, bind=engine)
else:
    # 使用MySQL数据库
    SQLALCHEMY_DATABASE_URL = f"mysql+pymysql://{config.UserName}:{config.Password}@{config.Host}:{config.Port}/{config.DataBase}"
    engine = create_engine(SQLALCHEMY_DATABASE_URL, pool_pre_ping=True, pool_size=5, pool_timeout=30, pool_recycle=1)
    SessionLocal = sessionmaker(autocommit=False, autoflush=True, bind=engine)
    session = scoped_session(SessionLocal)
Base = declarative_base()


def get_db():
    db = session
    try:
        yield db
    finally:
        db.remove()

在就是直接去调用session就可以了，使用完之后再remove

def test(task, db: scoped_session = session):
    db.query(ta).filter(ta.id == task.id).delete()
    db.add(ta(**task.to_dict()))
    db.commit()
    db.remove()

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

如有侵权请联系 cloudcommunity@tencent.com 删除

编程算法