SQLAlchemy使用教程（以SQLite为例）

1.  安装依赖

   pip3 install sqlalchemy==1.4.46

   pip install sqlacodegen

   如果是mysql则多安装一个依赖

   pip3 install pymysql

2. 在项目中新建一个db文件夹（后来：推荐将xxx.db文件放在启动文件同级目录，不要再新建文件夹了，后面就可以使用相对路径了，比较方便）
   

3. 在db文件夹中新建SQLite数据库（即新建一个以.db为后缀的空文件）【mysql跳过这一步】
   

4. 新建配置文件db_config.py
   其他数据连接的写法参考：SQLAlchemy创建各种数据库连接的_苍穹之跃的博客-CSDN博客

   import osimport platformfrom sqlalchemy import create_enginefrom sqlalchemy.orm import declarative_base, sessionmakerfrom contextlib import contextmanager"""Mysql连接"""# MYSQL_URL = f'mysql+pymysql://root:123456@localhost:3306/fast'"""Sqlite连接：注意注意注意：这个URI连接的相对地址，指的是相对于最外层调用的文件的相对位置，而不是此文件的相对位置。所以最好是使用绝对路径。"""# 获取当前文件的绝对路径SQLITE_URI = Noneif str(platform.system().lower()) == 'windows':    path = __file__.replace(fr"\{os.path.basename(__file__)}", "").replace("\\\\", "\\")    SQLITE_URI = fr'sqlite:///{path}\fast.db''?check_same_thread=False'    print(f'数据库路径：{SQLITE_URI}')elif str(platform.system().lower()) == 'linux':    path = __file__.replace(fr"/{os.path.basename(__file__)}", "").replace("//", "/")    SQLITE_URI = fr'sqlite:///{path}/fast.db''?check_same_thread=False'    print(f'数据库路径：{SQLITE_URI}')else:    print(f"未知系统：{platform.system().lower()}")# 操作数据句柄engine = create_engine(SQLITE_URI)Base = declarative_base(engine)DbSession = sessionmaker(bind=engine)# 这里一定要用上下文去管理session,否则会出现很多诡异的情况！！！切记db_session = DbSession()@contextmanagerdef session_maker(session=db_session):    try:        yield session        session.commit()    except:        session.rollback()        raise    finally:        session.close()# 逆向工程 自动生成模型文件if __name__ == '__main__':    os.system(f'sqlacodegen {SQLITE_URI} > models.py')

5. 用Navicat连接第三步创建的SQLite数据库，并新建业务表
   

6. 执行db_config文件中的逆向工程
   
   此时在db文件夹下会自动生成相应的模型文件：models.py，此文件最好不要手动修改。如果后面改动了表结构，再执行一次逆向工程即可。
   

7. ORM操作

   from sqlalchemy import funcfrom db.db_config import sessionfrom db.models import DatabaseConnectionif __name__ == '__main__':with session_maker() as session:    # ADD    session.add(DatabaseConnection(link_name='代码生成器', type='DAO', host='localhost', port='3306', username='admin',password='123456'))    session.commit()    # ADD_BATCH    session.add_all([        DatabaseConnection(link_name='代码生成器1', type='DAO', host='localhost', port='3306', username='admin',password='123456'),        DatabaseConnection(link_name='代码生成器2', type='DAO', host='localhost', port='3306', username='admin',password='123456'),        DatabaseConnection(link_name='代码生成器3', type='DAO', host='localhost', port='3306', username='admin',password='123456')    ])    session.commit()    # DELETE    session.query(DatabaseConnection).filter(DatabaseConnection.id == 6).delete()    session.commit()    # UPDATE    session.query(DatabaseConnection).filter(DatabaseConnection.id == 5).update({DatabaseConnection.link_name: '安特磁材2'})    session.commit()    # SELECT 精确查询    session.query(DatabaseConnection).filter(DatabaseConnection.host == 'localhost').all()    # SELECT 获取部分字段    session.query(DatabaseConnection.link_name).filter(DatabaseConnection.host == 'localhost').all()    # SELECT 模糊查询    session.query(DatabaseConnection).filter(DatabaseConnection.port.like('%33%')).all()    # SELECT 正则查询    session.query(DatabaseConnection).filter(DatabaseConnection.port.op("regexp")("[\d\D]*")).all()    # SELECT 排序    session.query(CodeTemplateGroup).order_by(CodeTemplateGroup.is_default.desc()).all()    # SELECT 统计查询    session.query(DatabaseConnection).filter(DatabaseConnection.port.op("regexp")("[\d\D]*")).count()    # SELECT 调用数据库内置函数    session.query(func.count(DatabaseConnection.link_name)).one()    session.query(func.count(DatabaseConnection.link_name)).one_or_none()        # 执行原生SQL    records = session.execute('select * from database_connection')    # 逻辑操作 not_取反，or_取或，and_取与    session.query(OperatingInstructionLog) \            .filter(OperatingInstructionLog.type == 2) \            .filter(            not_(or_(OperatingInstructionLog.open_time >= datetime(date.year, date.month, date.day + 1, 0, 0, 0),                     OperatingInstructionLog.close_time <= datetime(date.year, date.month, date.day, 0, 0, 0)))) \            .order_by(OperatingInstructionLog.open_time.asc())\            .all()

8.擦除对象

来源地址：https://blog.csdn.net/wenxingchen/article/details/127801084