devbox
小小林熬夜学编程
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

SQLAlchemy高级SqlHelper封装_sqlalchemy 封装

作者：小小林熬夜学编程 | 2024-05-17 18:37:12

赞

踩

sqlalchemy 封装

环境与版本

python: 3.10

SQLAlchemy: 2.0.9

网上好多sql语句查询相关的代码执行异常，不确定是否与版本有有关

说明

封装了比较实用的 复合唯一索引表的插入或更新，多条件查询，叠加条件查询，以及返回pandas DataFrame。

容易出错的问题

1. 执行sql语句错误提示：sqlalchemy.exc.ObjectNotExecutableError: Not an executable object‘......’

参考

2. pd.read_sql()  pd.read_sql_query() 报错 AttributeError: 'OptionEngine' object has no attribute 'execute'

read_sql() pd.read_sql_query()的con参数，必须是session.bind.connect() 而不是session.bind

数据库初始化以及Model定义

  1. from sqlalchemy import create_engine
  2. from sqlalchemy.orm import sessionmaker, relationship
  3. from sqlalchemy.ext.declarative import declarative_base
  4. from sqlalchemy import Column, Integer, String, ForeignKey, Index
  5.  
  6. # 数据库配置
  7. db_config = {
  8.     'sqlite': {
  9.         'engine': 'sqlite:///mydatabase.db'
  10.     },
  11.     'mysql': {
  12.         'engine': 'mysql+pymysql://username:password@host/database'
  13.     }
  14. }
  15.  
  16. # 根据配置选择 engine
  17. db_choice = 'sqlite'
  18. if db_choice == 'sqlite':
  19.     engine = create_engine(db_config['sqlite']['engine'])
  20. elif db_choice == 'mysql':
  21.     engine = create_engine(db_config['mysql']['engine'])
  22. else:
  23.     engine = create_engine('sqlite:///mydatabase.db')
  24.  
  25. Session = sessionmaker(bind=engine)
  26.  
  27. Base = declarative_base()
  28.  
  29.  
  30. class Address(Base):
  31.     __tablename__ = 'address'
  32.     id = Column(Integer, primary_key=True, autoincrement=True)
  33.     address = Column(String)
  34.  
  35.  
  36. class User(Base):
  37.     __tablename__ = 'users'
  38.  
  39.     id = Column(Integer, primary_key=True, autoincrement=True)
  40.     addr_id = Column(Integer, ForeignKey('address.id'))
  41.     name = Column(String)
  42.     age = Column(Integer)
  43.  
  44.     def __repr__(self):
  45.         return f"<User(id='{self.id}', name='{self.name}', age='{self.age}')>"
  46.  
  47.  
  48. class Author(Base):
  49.     __tablename__ = 'author'
  50.     id = Column(Integer, primary_key=True)
  51.     name = Column(String)
  52.     books = relationship('Book', back_populates='author')
  53.  
  54.  
  55. class Book(Base):
  56.     __tablename__ = 'book'
  57.     id = Column(Integer, primary_key=True)
  58.     title = Column(String)
  59.     author_id = Column(Integer, ForeignKey('author.id'))
  60.     author = relationship('Author', back_populates='books')
  61.  
  62.  
  63. class Department(Base):
  64.     __tablename__ = 'department'
  65.     id = Column(Integer, primary_key=True)
  66.     name = Column(String)
  67.     employees = relationship('Employee', back_populates='department')
  68.  
  69.  
  70. class Employee(Base):
  71.     __tablename__ = 'employee'
  72.     id = Column(Integer, primary_key=True)
  73.     name = Column(String)
  74.     remark = Column(String)
  75.     score = Column(Integer)
  76.     department_id = Column(Integer, ForeignKey('department.id'))
  77.     department = relationship('Department', back_populates='employees')
  78.  
  79.  
  80. class Code(Base):
  81.     __tablename__ = 'code'
  82.     id = Column(Integer, primary_key=True)
  83.     symbol = Column(String)
  84.     exchange = Column(String)
  85.     remark = Column(String)
  86.     # 复合唯一索引
  87.     __table_args__ = (
  88.         Index('ix_vt_symbol', 'symbol', 'exchange', unique=True),
  89.     )
  90.  
  91.  
  92. Base.metadata.create_all(engine)

SqlHelper类

  1. import pandas as pd
  2. import sqlalchemy
  3.  
  4. from sqlalchemy import or_, and_, text
  5. from sqlalchemy.exc import IntegrityError
  6.  
  7. from sqltool.models import Session, User, Address, Book, Author, Department, Employee, Code
  8.  
  9.  
  10. def or_filters(*args):
  11.     return or_(*args)
  12.  
  13.  
  14. def and_filters(*args):
  15.     return and_(*args)
  16.  
  17.  
  18. class SqlHelper:
  19.     def __init__(self):
  20.         self.session = Session()
  21.  
  22.     def add(self, obj):
  23.         self.session.add(obj)
  24.         self.session.commit()
  25.  
  26.     def delete(self, obj):
  27.         self.session.delete(obj)
  28.         self.session.commit()
  29.  
  30.     def update(self, obj):
  31.         self.session.merge(obj)
  32.         self.session.commit()
  33.  
  34.     def upsert(self, model):
  35.         self._upsert(model)
  36.         self.session.commit()
  37.  
  38.     def _upsert(self, model):
  39.         '''
  40.         根据唯一索引查找记录，再决定更新(补全primary_key的自增列)或新增记录
  41.         '''
  42.         unique = self.get_unique_keys(model)
  43.         obj = self.session.query(model.__class__).filter(
  44.             # getattr(model.__class__, primary_key) == getattr(model, primary_key),
  45.             *[(getattr(model.__class__, c) == getattr(model, c)) for c in unique]
  46.         ).first()
  47.         if obj:
  48.             primary_key = self.get_primary_key(model)
  49.             val = getattr(obj, primary_key)
  50.             setattr(model, primary_key, val)
  51.             self.session.merge(model)
  52.         else:
  53.             self.session.add(model)
  54.  
  55.     def add_all(self, models):
  56.         try:
  57.             self.session.begin()
  58.             self.session.add_all(models)
  59.             self.session.commit()
  60.         except IntegrityError:
  61.             # 如果出现唯一键冲突错误，回滚事务
  62.             self.session.rollback()
  63.  
  64.     def update_all(self, models):
  65.         try:
  66.             for m in models:
  67.                 self.session.merge(m)
  68.             self.session.commit()
  69.         except IntegrityError:
  70.             self.session.rollback()
  71.  
  72.     def delete_all(self, models):
  73.         try:
  74.             for m in models:
  75.                 self.session.delete(m)
  76.             self.session.commit()
  77.         except IntegrityError:
  78.             self.session.rollback()
  79.  
  80.     def upsert_all(self, models):
  81.         try:
  82.             self.session.begin()
  83.             for m in models:
  84.                 self._upsert(m)
  85.             self.session.commit()
  86.         except IntegrityError:
  87.             self.session.rollback()
  88.  
  89.     def query(self, model):
  90.         return self.session.query(model)
  91.  
  92.     def update_by(self, model, filters, col_value_map):
  93.         return self.session.query(model).filter(filters).update(col_value_map)
  94.  
  95.     def query_by(self, model, *args):
  96.         return self.session.query(model).filter(and_(*args))
  97.  
  98.     def query_or(self, model, *args):
  99.         return self.session.query(model).filter(or_(*args))
  100.  
  101.     def query_by_func(self, model, func):
  102.         '''
  103.         使用 or_filters 和 and_filters 叠加查询条件
  104.         demo
  105.             and_filter(or_filters(User.age < 30, User.age > 60), User.name == 'lisi_6')
  106.             等同于
  107.             (User.age < 30 or User.age > 60) and User.name == 'lisi_6'
  108.         '''
  109.         return self.session.query(model).filter(func)
  110.  
  111.     def get_by(self, model, **kwargs):
  112.         return self.session.query(model).filter_by(**kwargs).first()
  113.  
  114.     def count_by(self, model, **kwargs):
  115.         return self.session.query(model).filter_by(**kwargs).count()
  116.  
  117.     def exists(self, model, **kwargs):
  118.         return self.session.query(model).filter_by(**kwargs).exists()
  119.  
  120.     def join(self, model1, model2):
  121.         return self.session.query(model1).join(model2)
  122.  
  123.     def join_ex(self, model1, model2):
  124.         return self.session.query(model1, model2)
  125.  
  126.     def first(self, query):
  127.         return query.first()
  128.  
  129.     def commit(self):
  130.         self.session.commit()
  131.  
  132.     def close(self):
  133.         self.session.commit()
  134.         self.session.close()
  135.  
  136.     def get_primary_key(self, model):
  137.         for col in model.__table__.columns:
  138.             if col.primary_key:
  139.                 return col.name
  140.  
  141.     def get_unique_keys(self, model) -> list[str]:
  142.         ''' 获取唯一键列 '''
  143.         # 1.从列中unique属性获取列名
  144.         unique = [c.name for c in model.__table__.columns if c.unique]
  145.         # 2.从__table_args__自定义索引(Index)中获取unique的列名
  146.         idxs = list(model.__table__.indexes)
  147.         for idx in idxs:
  148.             if not idx.unique:
  149.                 continue
  150.             for c in idx.columns:
  151.                 unique.append(c.name)
  152.         return unique
  153.  
  154.     def execute_sql(self, sqlstr) -> list[tuple]:
  155.         return self.session.execute(text(sqlstr)).fetchall()
  156.  
  157.     def read_from(self, sql) -> pd.DataFrame:
  158.         '''
  159.             查询sql语句，必须使用text()
  160.             pandas 需要connection，不是engine
  161.         '''
  162.         return pd.read_sql_query(text(sql), self.session.bind.connect())
  163.  
  164.     def read_model(self, model) -> pd.DataFrame:
  165.         '''
  166.             查询sql语句，必须使用text()
  167.             pandas 需要connection，不是engine
  168.         '''
  169.         return pd.read_sql(self.session.query(model).statement, self.session.bind.connect())

demo

  1.  
  2. def demo():
  3.     sql_helper = SqlHelper()
  4.     count = sql_helper.count_by(Address)
  5.     addr0 = Address()
  6.     addr0.address = f"浙江杭州_{count + 1}"
  7.     sql_helper.add(addr0)
  8.     # addr = sql_helper.query(Address).first()
  9.     user = User()
  10.     user.addr_id = addr0.id
  11.     user.age = 33
  12.     user.name = f"lisi_{count + 1}"
  13.     sql_helper.add(user)
  14.  
  15.     users = sql_helper.query_by(User, User.name == 'Tom').all()
  16.     users = sql_helper.query(User).filter(User.age > 30).all()
  17.     users = sql_helper.query(User).filter(User.name.like('a%')).all()
  18.     tom = sql_helper.get_by(User, name='Tom')
  19.     count = sql_helper.count_by(User)
  20.     exists = sql_helper.exists(User, age=30)
  21.     first_user = sql_helper.first(sql_helper.query(User))
  22.     query_result = sql_helper.join_ex(Address, User).filter(User.addr_id == Address.id).all()
  23.     for address, user in query_result:
  24.         print(user.id, user.name, address.id, address.address)
  25.     # query_result = sql_helper.query(User).join(Address).filter(User.addr_id == Address.id).all()
  26.     query_result = sql_helper.query(User).join(Address).all()
  27.     for result in query_result:
  28.         print(result)
  29.  
  30.     # 连接
  31.     author1 = Author(name='Jam')
  32.     author2 = Author(name='Sam')
  33.     sql_helper.add(author1)
  34.     sql_helper.add(author2)
  35.     book1 = Book(title="C Language", author_id=author1.id)
  36.     book2 = Book(title="C++ Language", author_id=author1.id)
  37.     book3 = Book(title="Python", author_id=author2.id)
  38.     book4 = Book(title="JavaScript", author_id=author2.id)
  39.     sql_helper.add(book1)
  40.     sql_helper.add(book2)
  41.     sql_helper.add(book3)
  42.     sql_helper.add(book4)
  43.  
  44.     # n--1 连接
  45.     books = sql_helper.query(Book).join(Author).all()
  46.     for book in books:
  47.         print(f'Book: {book.title} Author:{book.author.name}')
  48.  
  49.     authors = sql_helper.query(Author).join(Book).all()
  50.     for author in authors:
  51.         for book in author.books:
  52.             print(f'Author: {author.name} Book: {book.title}')
  53.  
  54.     # 1--n 连接
  55.     department = Department(name='Sales')
  56.     sql_helper.add(department)
  57.     employee1 = Employee(name='John', department=department, remark='test1111')
  58.     employee2 = Employee(name='Jane', department=department, remark='test2222')
  59.     sql_helper.add(employee1)
  60.     sql_helper.add(employee2)
  61.     dept = sql_helper.query(Department).filter_by(name='Sales').first()
  62.     employees = dept.employees
  63.     for employee in employees:
  64.         print(employee.name)
  65.  
  66.     # 单条件批量更新
  67.     rst = sql_helper.update_by(Employee, Employee.name == 'John', {Employee.remark: employee1.remark})
  68.     sql_helper.commit()
  69.     # 多条件批量更新
  70.     rst = sql_helper.update_by(Employee, or_filters(Employee.name == 'John', Employee.name == 'Jane'),
  71.                                {Employee.remark: employee2.remark})
  72.     sql_helper.commit()
  73.     print('------------------------------------------------------------')
  74.     # 复合唯一索引的存在更新否则插入
  75.     code1 = Code(symbol='SA309', exchange='CZCE', remark='test11111')
  76.     code2 = Code(symbol='SA309', exchange='CZCE', remark='test22222')
  77.     # sql_helper.add(code1)
  78.     sql_helper.upsert(code1)
  79.     sql_helper.upsert(code2)
  80.     print('------------------------------------------------------------')
  81.     # or 查询
  82.     users = sql_helper.query_or(User, User.age > 55, User.age < 30, User.age == 44, User.name == 'lisi_40').all()
  83.     for user in users:
  84.         print(user)
  85.     print('------------------------------------------------------------')
  86.     # and 查询
  87.     users = sql_helper.query_by(User, User.name == 'lisi_6', User.age > 45)
  88.     for user in users:
  89.         print(user)
  90.     print('------------------------------------------------------------')
  91.     # 多条件嵌套查询 or + and
  92.     users = sql_helper.query_by_func(User, and_filters(or_filters(User.age < 30, User.age > 60),
  93.                                                       User.name == 'lisi_6')).all()
  94.     for user in users:
  95.         print(user)
  96.     print('------------------------------------------------------------')
  97.     result = sql_helper.execute_sql('select * from users;')
  98.     for item in result:
  99.         print(item)
  100.     print('------------------------------------------------------------')
  101.     df = sql_helper.read_model(User)
  102.     print(df.head())
  103.     print('------------------------------------------------------------')
  104.     df = sql_helper.read_from('select * from users;')
  105.     print(df.tail())
  106.     print('------------------------------------------------------------')
  107.     result = sql_helper.execute_sql('select * from employee;')
  108.     for item in result:
  109.         print(item)
  110.     print('------------------------------------------------------------')
  111.     sql_helper.close()
  112.  
  113.  
  114. if __name__ == '__main__':
  115.     print(sqlalchemy.__version__)
  116.     demo()

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小小林熬夜学编程/article/detail/584782
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号