SQLAlchemy是Python的ORM库，支持多种数据库。
建立连接
from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://user:pass@localhost/db?charset=utf8mb4',
echo=True, pool_recycle=7200, pool_size=5, max_overflow=10, pool_timeout=30)
create_engine第一个参数是database url，用到了pymysql驱动。
参数echo用于打印执行中的关键日志，包括SQL执行语句。
pool_recycle表示connection空闲7200秒就会重新获取。
pool_size指定连接池大小。
max_overflow指定超过连接池大小的最大数。
pool_timeout获取连接的超时阈值。
poolclass设置为NullPool，禁用连接池，session.Close()之后会立即断开数据库连接。
2013, 'Lost connection to MySQL server during query
([Errno 104] Connection reset by peer)
设置 pool_pre_ping=True和 pool_recycle可以每次执行都查看连接是否可用。
pool_recycle的值 应该比mysql中设置的 interactive_timeout 和 wait_timeout 值小才有效。
pool_timeout的值(默认10s) 应该比 mysql的 connect_timeout值 小或者等于。
SHOW GLOBAL VARIABLES LIKE '%timeout%';
用 urllib.parse.quote_plus编码@特殊字符。
from urllib import parse
from sqlalchemy import create_engine
user = "root"
password = "******@108"
host = "127.0.0.1"
db = "db_name"
pwd = parse.quote_plus(password)
engine = create_engine(f"mysql+pymysql://{user}:{pwd}@{host}:3306/{db}?charset=utf8")
建立映射
from sqlalchemy.sql.schema import Column
from sqlalchemy.types import BigInteger, String, Integer
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(BigInteger, primary_key=True)
name = Column(String(255))
age = Column(Integer)
gender = Column(String(255))
另外可以使用sqlacodegen（pip install sqlacodegen）自动生成model：
sqlacodegen mysql+pymysql://user:password@localhost/test --outfile=models.py --tables users
from sqlalchemy import Column, String
from sqlalchemy.dialects.mysql import BIGINT, INTEGER
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
metadata = Base.metadata
class User(Base):
__tablename__ = 'users'
id = Column(BIGINT(20), primary_key=True)
name = Column(String(255, 'utf8_unicode_ci'))
age = Column(INTEGER(11))
gender = Column(String(255, 'utf8_unicode_ci'))
参考类型：
类型介绍Integer/BigInteger/SmallInteger整型Boolean布尔类型. Python 中表现为 True/False , 数据库根据支持情况, 表现为 BOOLEAN 或 SMALLINT . 实例化时可以指定是否创建约束(默认创建).Date/DateTime/Time (timezone=False)日期类型, Time 和 DateTime 实例化时可以指定是否带时区信息Interval时间偏差类型. 在 Python 中表现为 datetime.timedelta() , 数据库不支持此类型则存为日期Enum (enums, *kw)枚举类型, 根据数据库支持情况, SQLAlchemy 会使用原生支持或者使用 VARCHAR 类型附加约束的方式实现. 原生支持中涉及新类型创建, 细节在实例化时控制Float浮点小数Numeric (precision=None, scale=None, decimal_return_scale=None, …)定点小数, Python 中表现为 DecimalLargeBinary (length=None)字节数据. 根据数据库实现, 在实例化时可能需要指定大小PickleTypePython 对象的序列化类型String (length=None, collation=None, …)字符串类型, Python 中表现为 Unicode , 数据库表现为 VARCHAR , 通常都需要指定长度Unicode类似与字符串类型, 在某些数据库实现下, 会明确表示支持非 ASCII 字符. 同时输入输出也强制是 Unicode 类型Text长文本类型, Python 表现为 Unicode , 数据库表现为 TEXTUnicodeText参考 Unicode
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql.schema import Table
Base = declarative_base()
metadata = Base.metadata
metadata.bind = engine
class User(Base):
__table__ = Table('users', metadata, autoload=True)
设置autoload就可以自动生成映射，另外Base.metadata需要提前绑定engine。
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
session = Session()
query = session.query(User).filter(User.age >= 19).filter(User.name.like("王%"))
for user in query.all():
print(user.age)
查询用到了Session对象，类似一个容器，容器里是 identity map 的结构对象。
用filter过滤数据。
session.query(Account).filter(Account.user_name=='Mark')
session.query(Account).filter(Account.salary==2000)
session.query(Account).filter(Account.user_name !="mark" )
session.query(Account).filter(Account.salary !=2000)
session.query(Account).filter(Account.salary >2000)
session.query(Account).filter(Account.salary <2000)
session.query(Account).filter(Account.salary 2000)
session.query(Account).filter(Account.salary >=2000)
session.query(Account).filter(Account.user_name.like('%i%'))
session.query(Account).filter(Account.title.like('%Manager'))
session.query(Account).filter(Account.user_name.like('Da%'))
session.query(Account).filter(~Account.id.in_([1,3,5]))
session.query(Account).filter(~Account.id.in_([2000,3000,4000]))
session.query(Account).filter(~Account.title.in_(['Accountant','Engineer']))
session.query(Account).filter(Account.salary==None)
session.query(Account).filter(Account,salary.is_(None))
session.query(Account).filter(Account.salary!=None)
session.query(Account).filter(Account.salary.isnot(None))
session.query(Account).filter(~Account.id.in_([1,3,5]))
session.query(Account).filter(~Account.id.in_([2000,3000,4000]))
session.query(Account).filter(~Account.title.in(['Accountant','Engineer']))
session.query(Account).filter(Account.title=='Engineer',Account.salary=3000)
from sqlalchemy import and_
session.query(Account).filter(and_(Account.title=='Engineer',Account.salary=3000))
session.query(Account).filter(Account.title=='Engineer').filter(Account.salary=3000)
from sqlalchemy import or_
session.query(Account).filter(or_(Account.title=='Engineer',Account.salary=3000))
user1 = User()
user1.name = "赵六"
user1.age = 21
user1.gender = "女"
session.add(user1)
session.commit()
先建立一个User实例，然后调用session的add方法，最后commit。
批量插入新增
session = Session()
objects = [
User(name="u1"),
User(name="u2"),
User(name="u3")]
session.bulk_save_objects(objects)
session = Session()
objects = [User(name="u1"), User(name="u2"), User(name="u3")]
session.add_all(objects)
session.commit()
user2 = session.query(User).filter_by(name='赵六').first()
user2.gender = "男"
session.commit()
先查找到User实例，然后修改实例属性，最后调用session的commit修改数据库。
user3 = session.query(User).filter_by(name='赵六').first()
session.delete(user3)
session.commit()
多表查询
Session = sessionmaker(bind=engine)
session = Session()
query = session.query(User, Address).join(Address, User.id == Address.user_id)
for user in query.all():
print(user[0].name)
print(user[1].add)
Original: https://blog.csdn.net/lilongsy/article/details/128430737
Author: 小龙在山东
Title: SQLAlchemy连接MySQL及记录的查询、更新、删除、多表关联查询
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/786957/
转载文章受原作者版权保护。转载请注明原作者出处！