python sqlalchemy-orm-sqlite

使用老版sqlalchemy - orm sqllite执行demo程序 极简SQLAlchemy原生SQL Demo 规则所有SQL语句由调用者完整手写Navicat可直接运行无任何占位符? / :xxx 仅提供通用执行工具内部不拼接任何SQL from datetime import datetime from pathlib import Path from sqlalchemy import ( create_engine, Column, String, Text, Integer, Float, DateTime, JSON, text ) from sqlalchemy.orm import declarative_base, sessionmaker, Session # # 1. 引擎 会话不变 # DB_PATH Path(__file__).parent / orm_demo.db DATABASE_URL fsqlite:///{DB_PATH} engine create_engine( DATABASE_URL, connect_args{check_same_thread: False}, echoFalse, ) 可能是一个session工厂的function SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) Base declarative_base() # # 2. 模型仅用于初始化建表业务不用ORM # class Article(Base): 仅初始化建表使用业务全部手写完整SQL __tablename__ articles id Column(Integer, primary_keyTrue, autoincrementTrue) title Column(String(200), nullableFalse) content Column(Text, default) author Column(String(100), default匿名) word_count Column(Integer, default0) rating Column(Float, default0.0) tags Column(JSON, nullableTrue) status Column(String(20), defaultdraft) created_at Column(DateTime, defaultdatetime.now) updated_at Column(DateTime, defaultdatetime.now, onupdatedatetime.now) # # 3. 极简通用工具函数核心无任何SQL拼接逻辑 # def get_session() - Session: return SessionLocal() def init_db(): 初始化数据库清空重建表 if DB_PATH.exists(): 直接删除 DB_PATH.unlink() Base.metadata.create_all(bindengine) print(f[INFO] 数据库初始化完成: {DB_PATH}) def fetch_one(full_sql: str): 执行完整查询SQL返回单条字典 :param full_sql: 调用者手写完整可直接在Navicat运行的SQL session get_session() try: sql text(full_sql) res session.execute(sql) row res.fetchone() return row._asdict() if row else None finally: session.close() def fetch_all(full_sql: str): 执行完整查询SQL返回全部结果列表 :param full_sql: 调用者手写完整可直接在Navicat运行的SQL session get_session() try: sql text(full_sql) res session.execute(sql) return [row._asdict() for row in res.fetchall()] finally: session.close() def execute_commit(full_sql: str) - int: 执行INSERT / UPDATE / DELETE完整SQL自动提交返回受影响行数 :param full_sql: 调用者手写完整可直接在Navicat运行的SQL session get_session() try: sql text(full_sql) res session.execute(sql) session.commit() return res.rowcount finally: session.close() # # 4. 事务通用工具多条完整SQL同一事务 # def transaction_execute(sql_list: list[str]): 事务执行多条完整SQL全部成功才提交任意异常全部回滚 :param sql_list: 完整SQL字符串列表每条都能单独在Navicat运行 session get_session() try: for sql in sql_list: session.execute(text(sql)) session.commit() print(✅ 事务全部执行成功已提交) return True except Exception as e: session.rollback() print(f 事务异常全部回滚: {e}) return False finally: session.close() # # 5. 调用示例所有SQL均为完整可直接Navicat运行 # def main(): print( * 60) print(极简原生SQL Demo - 全部SQL由调用者完整编写) print( * 60) init_db() # -------------------------- # 新增 - 完整INSERT SQL无占位符 # -------------------------- print(\n--- 新增文章完整SQL---) now datetime.now().isoformat() # 调用者自行拼接完整SQLNavicat直接执行 insert_sql f INSERT INTO articles (title, content, author, word_count, rating, tags, status, created_at, updated_at) VALUES (SQLAlchemy极简原生SQL, 全部SQL手写完整语句无任何占位符, 张三, 1200, 4.5, [Python,SQLAlchemy], published, {now}, {now}) execute_commit(insert_sql) insert_sql2 f INSERT INTO articles (title, content, author, word_count, rating, tags, status, created_at, updated_at) VALUES (FastAPI开发笔记, Web后端框架实战, 张三, 2500, 4.8, [Python,FastAPI], published, {now}, {now}) execute_commit(insert_sql2) # -------------------------- # 查询单条 # -------------------------- print(\n--- 根据ID查询单条 ---) select_one_sql SELECT * FROM articles WHERE id 1; article fetch_one(select_one_sql) print(f查询结果{article[title]}) # -------------------------- # 多条件分页查询完整SQL # -------------------------- print(\n--- 多条件分页查询 作者张三已发布 ---) select_page_sql SELECT * FROM articles WHERE author 张三 AND status published ORDER BY created_at DESC LIMIT 10 OFFSET 0; article_list fetch_all(select_page_sql) for item in article_list: print(f[{item[id]}] {item[title]}) # -------------------------- # 更新操作 完整UPDATE SQL # -------------------------- print(\n--- 更新文章评分 ---) update_sql UPDATE articles SET rating 5.0, updated_at 2026-07-26 14:30:00 WHERE id 1; execute_commit(update_sql) # -------------------------- # 删除操作 完整DELETE SQL # -------------------------- print(\n--- 创建临时文章并删除 ---) temp_insert_sql INSERT INTO articles (title, content, author, status) VALUES (临时待删文章, 测试删除, 测试, draft); execute_commit(temp_insert_sql) delete_sql DELETE FROM articles WHERE title 临时待删文章; execute_commit(delete_sql) # -------------------------- # 事务演示多条完整SQL一条异常全部回滚 # -------------------------- print(\n--- 事务回滚演示 ---) # 第二条SQL故意写错字段触发异常回滚 sql_batch [ INSERT INTO articles (title, author, status) VALUES (事务测试1, 测试, draft);, INSERT INTO articles (错误字段) VALUES (事务测试2);, ] transaction_execute(sql_batch) # -------------------------- # 最终全部数据查询 # -------------------------- print(\n--- 数据库全部文章 ---) all_sql SELECT id, title, author, rating, status FROM articles ORDER BY id; final_data fetch_all(all_sql) for row in final_data: print(f[{row[id]}] {row[title]} | 作者:{row[author]} | 评分:{row[rating]}) print(\n Demo 执行完毕) if __name__ __main__: main()