SQLAlchemy攻略

SQLAlchemy基础教程

• SQLAlchemy是一个基于Python的ORM框架。该框架是建立在DB-API之上，使用关系对象映射进行数据库操作。

安装

pip install sqlalchemy

连接数据库

• 由于SQLAlchemy本身无法操作数据库，因此需要依赖第三方模块，遵循DB-API规范。

• 以下是不同数据库的API

  # MySQL-PYthon
  mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
  
  #pymysql
  mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
  
  # MySQL-Connector
  mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
  
  # cx_Oracle
  oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
  

• 连接数据库

  from sqlalchemy import create_engine
  
  # create_engine就是建立连接
  conn = create_engine(
      "mysql+pymysql://root:123@182.92.149.42:3306/数据库名?charset=utf8mb4",
      max_overflow=0,   # 超过连接池大小外最多创建的连接数
      pool_size=5,      # 连接池大小
      pool_timeout=30,  # 连接池中没有线程最多等待时间，否则报错
      pool_recycle=-1,  # 多久之后对连接池中的连接进行回收（重置）-1不回收
  )
  

执行原生SQL

from sqlalchemy import create_engine

conn = create_engine(
    "mysql+pymysql://root:123@182.92.149.42:3306/数据库名?charset=utf8mb4",
    max_overflow=0,   # 超过连接池大小外最多创建的连接数
    pool_size=5,      # 连接池大小
    pool_timeout=30,  # 连接池中没有线程最多等待时间，否则报错
    pool_recycle=-1,  # 多久之后对连接池中的连接进行回收（重置）-1不回收
)

def test():
  ret = conn.execute("select * from MyTest")
  result = ret.fetchall()
  print(result)
  ret.close()
  
if __name__ =  '__main__':
  test()

ORM

重要参数：

1. __tablename__ 表名
2. __table_args__ 在此处添加约束，例如Index索引或联合索引，UniqueConstraint联合唯一
3. ForeignKey的字段的建立，需要指定外键绑定哪个表的哪个字段
4. relationship 不生成表结构，第一个参数是关联到哪个类(表), backref是给关联的那个类反向查询用的

必继承：

1. Base = declarative_base()，后面所有的表创建的时候都要继承这个Base类

from sqlalchemy import create_engine, ForeignKey, UniqueConstraint, Index
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.orm import relationship
from sqlalchemy import Index, UniqueConstraint

conn = create_engine(
    "mysql+pymysql://root:123abc@127.0.0.1:3306/mytest?charset=utf8mb4",
    max_overflow=0,  # 超过连接池大小外最多创建的连接数
    pool_size=5,  # 连接池大小
    pool_timeout=30,  # 连接池中没有线程最多等待时间，否则报错
    pool_recycle=-1,  # 多久之后对连接池中的连接进行回收（重置）-1不回收
)

Base = declarative_base()


class Book(Base):
    __tablename__ = 'book'

    id = Column(Integer, primary_key=True)
    title = Column(String(64), nullable=False)
    publisher_id = Column(Integer, ForeignKey('publisher.id'))  # Book表的publisher_id
    publisher = relationship('Publisher', backref='books')  # 指定外键关联的对象 和 反向查询命名
    tags = relationship('Tag', backref='books', secondary='book2tag')  # 多对多，还需要secondary参数，创建第三张表

    __table_args__ = (
        # UniqueConstraint联合唯一，这个联合唯一的字段名为：uni_id_name
        UniqueConstraint("id", "title", name="uni_id_title"),
        # 联合索引
        Index("id", "title")
    )

    def __repr__(self):
        return self.title


class Publisher(Base):
    __tablename__ = 'publisher'

    id = Column(Integer, primary_key=True)
    title = Column(String(64), nullable=False)

    def __repr__(self):
        return self.title


class Tag(Base):
    __tablename__ = 'tag'

    id = Column(Integer, primary_key=True)
    title = Column(String(64), nullable=False)

    def __repr__(self):
        return self.title


class Book2Tag(Base):
    __tablename__ = 'book2tag'

    id = Column(Integer, primary_key=True)
    book_id = Column(Integer, ForeignKey('book.id'))
    tag_id = Column(Integer, ForeignKey('tag.id'))


def create_db():
    # metadata.create_all创建所有表
    Base.metadata.create_all(conn)


def drop_db():
    # metadata.drop_all删除所有表
    Base.metadata.drop_all(conn)

    
# 每次执行数据库操作的时候，都需要创建一个session,相当于管理器(相当于Django的ORM的objects)
session_factory = sessionmaker(bind=conn)
# 线程安全，基于本地线程实现每个线程用同一个session
Session = scoped_session(session_factory)
# 实例化(相当于实现了一个单例模式)
session = Session()
# session2 = Session() --> session is session2

if __name__ = 'main':
  create_db()  # 此时数据库表结构已搭好

重头戏：查询

查询关键字如下：

1. filter
2. filter_by
3. and_
4. or_
5. join 内联
6. outerjoin 外联
7. first()
8. all()

单表查询

# 查询id为2的书
session.query(Book).filter_by(id=2).first()
# 不加first() 返回的是query对象 <class 'sqlalchemy.orm.query.Query'> ，需要通过frist()转换成model对象
# 如果是all()，那么返回列表

一对多查询

# 查询上海出版社出过哪些书
# 反向：
session.query(Publisher).filter_by(title='上海出版社').first().books  # 最后这个books就是别名，通过反向查询使用
# 正向：
session.query(Book).filter_by(publisher=session.query(Publisher).filter_by(title='上海出版社').first()).all()

多对多查询

# 查询golang这个标签属于哪些书
# 反向：
book = session.query(Tag).filter_by(title="golang").first().books

连表查

session.query(Book, Publisher).filter(Book.publisher_id == Publisher.id).all()
session.query(Book).join(Publisher).all()

组合条件 or_和 and_

# 查询id>=1,并且书名中含有"如"的书
session.query(Book).filter_by(and_(id>=1, Book.title.like('%如%')))