原生Python如何使用sqlalchemy框架操作mysql【web-api可以用flask-sqlalchemy比较简单】而用Python写客户端需要自己封装sqlalchemy

短命女 2021-08-13 18:58 330阅读 0赞

**常见错误：Table '数据库名称.表名称' doesn't exist**

**sqlalchemy.exc.ProgrammingError: (mysql.connector.errors.ProgrammingError) 1146 (42S02): Table '数据库名称.表名称' doesn't exist**

首先声明一点：关于原生sqlalchemy的使用建议直接看官方文档，以官方文档为准！【百度上很多其他文章未必100%正确，我在使用中就发现了一个现象】

比如这篇文章：[http://www.dba.cn/book/python3/FangWenShuJuKu/ShiYongSQLALCHEMY.html][http_www.dba.cn_book_python3_FangWenShuJuKu_ShiYongSQLALCHEMY.html]

**一直没有出现创建表的语句【create\_all(self.engine)】所以怎么都创建不成功，报错某某表不存在：**

### **【非常重要】必须把所有的Model全部导入一个共同的目录的\_\_init\_\_.py文件** ###

**   ** **创建所有的mysql表Table【在所有表创建完成之后,第一次必须执行此方法来创建所有的表】 **

文件位置：models.__init__.py文件
    
    from app.models.common.base import db
    from app.models.exchanges.okex.spot.okex_v3_spot_api_model import *
    
    """
        功能：统一初始化创建所有的数据库Model表
        【非常重要】必须把所有的Model全部导入一个共同的目录的__init__.py文件
        创建所有的mysql表Table【在所有表创建完成之后,第一次必须执行此方法来创建所有的表】
        **非常非常重要**：每个Model第一次初始化必须调用此方法创建Table,否则会报错【Table '数据库名称.表名称' doesn't exist】
        最佳位置：把所有的Model放入一个__init__.py文件初始化,统一调用一次此方法！
    """
    
    db.create_create_all_table()

官方手册：[https://docs.sqlalchemy.org/en/13/orm/tutorial.html][https_docs.sqlalchemy.org_en_13_orm_tutorial.html]

### **分享一个自己封装的案例：** ###

from sqlalchemy import create_engine
    from sqlalchemy.sql.schema import Column
    from sqlalchemy.sql.sqltypes import String, Integer, Text, DateTime, Float, Boolean
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy import orm
    from sqlalchemy.ext.declarative import declarative_base
    from contextlib import contextmanager
    from app.libs.common.class_iterator import ClassIterator
    from app.exchanges.utils.util_datatime import UtilsDatatime
    from app.exchanges.utils.util_global import UtilGlobal
    from app.libs.mysql.own_column_type import OwnUnsignedDecimal
    from app.config.secure import *
    
    
    class OwnQuery(orm.query.Query):
        # 重写父类原生SQLalchemy.orm.query.Query中的filter_by方法！！！
        # 数据库Base模型中,status=1,表面这条数据是真实有效的(非软删除)
        def filter_by(self, **kwargs):
            if "status" not in kwargs.keys():
                kwargs.update(status=1)
            return super().filter_by(**kwargs)
    
    
    class OwnSQLAlchemy:
        def __init__(self, query_class=None, **kwargs):
            # 初始化数据库连接:
            # uri格式为"mysql+mysqlconnector://用户名:密码@localhost:3306/数据库名称"
            self.engine = create_engine(SQLALCHEMY_DATABASE_URI)
            # 创建对象的基类:
            self.BaseModel = declarative_base()
            self.session = self.create_session(query_class=query_class, **kwargs)
            self.Column = Column
            # 常用数据类型快捷键
            self.String = String
            self.Integer = Integer
            self.Decimal = OwnUnsignedDecimal
            self.Text = Text
            self.DateTime = DateTime
            self.Float = Float
            self.Integer = Integer
            self.Boolean = Boolean
    
        def create_create_all_table(self):
            """
                功能：创建所有的mysql表Table【在所有表创建完成之后,第一次必须执行此方法来创建所有的表】
                **非常非常重要**：每个Model第一次初始化必须调用此方法创建Table,否则会报错【Table '数据库名称.表名称' doesn't exist】
                最佳位置：把所有的Model放入一个__init__.py文件初始化,统一调用一次此方法！
            """
            self.BaseModel.metadata.create_all(bind=self.engine)
    
        def create_session(self, query_class=None, **kwargs):
            """
                功能：数据库的增删改查【CRUD】都使用Session的实例对象
                特殊说明：如何重写sqlalchemy内部的结构,比如Query类
                    query_cls参数默认值为【sqlalchemy.orm.query.Query】具体参数请参考sessionmaker方法的官方文档
                    如何要修改query_cls的值,可以继承sqlalchemy.orm.query.Query重写内部方法即可
                参考官方文档：
                    1、https://docs.sqlalchemy.org/en/13/orm/session_api.html?highlight=sessionmaker#sqlalchemy.orm.session.sessionmaker
                    2、https://docs.sqlalchemy.org/en/13/orm/session_api.html?highlight=sessionmaker#sqlalchemy.orm.session.Session
            """
            if query_class is None:
                Session = sessionmaker(bind=self.engine, **kwargs)
            else:
                Session = sessionmaker(bind=self.engine, query_cls=query_class, **kwargs)
            self.session = Session()
            return self.session
    
        @contextmanager
        def auto_commit_db(self):
            """
                功能：利用contextmanager管理器,对try/except语句封装，使用的时候必须和with结合！！！
                补充：这里如果使用装饰器封装全局封装try-except也是可以的！
            """
            try:
                yield
                # 提交即保存到数据库:
                self.session.commit()
            except Exception as e:
                # 加入数据库commit提交失败，必须回滚！！！
                self.session.rollback()
                raise e
    
    
    db = OwnSQLAlchemy(query_class=OwnQuery)
    
    
    class Base(db.Model):
    pass
    
    
    后续所有的表都继承Base即可

###      ###

[http_www.dba.cn_book_python3_FangWenShuJuKu_ShiYongSQLALCHEMY.html]: http://www.dba.cn/book/python3/FangWenShuJuKu/ShiYongSQLALCHEMY.html
[https_docs.sqlalchemy.org_en_13_orm_tutorial.html]: https://docs.sqlalchemy.org/en/13/orm/tutorial.html