當前位置：首頁 > 运维知识 > 数据库 >内容正文

数据库

Flask 第三方组件之 SQLAlchemy

發布時間：2023/12/20 数据库 39 豆豆

生活随笔收集整理的這篇文章主要介紹了 Flask 第三方组件之 SQLAlchemy 小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

一、介紹

SQLAlchemy是一個基于Python實現的ORM框架。該框架建立在 DB API之上，使用關系對象映射進行數據庫操作，簡言之便是：將類和對象轉換成SQL，然后使用數據API執行SQL并獲取執行結果。

安裝：pip3 install sqlalchemy

組成部分：

Engine，框架的引擎
Connection Pooling ，數據庫連接池
Dialect，選擇連接數據庫的DB API種類
Schema/Types，架構和類型
SQL Exprression Language，SQL表達式語言

SQLAlchemy本身無法操作數據庫，其必須以來pymsql等第三方插件，Dialect用于和數據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...]更多：http://docs.sqlalchemy.org/en/latest/dialects/index.html

二、?使用

2.1??創建數據庫表

2.1.1 數據類型和數據庫的對應關系

類型名MySQL類型python類型描述

SmallInteger	smallint	int	取值范圍較小，一般為16位
Integer	int	int	普通整數，一般32位
BigInteger	bigint	int/long	不限精度的整數
Float	float	float	浮點數
Numeric	decimal	decimal.Decimal	定點數
String	varchar	str	變長字符串
Text	tinytext	str	變長字符串，64K,216?1216?1 2^{16}-1216?1=65535bytes
Text(65536)	mediumtext	str	變長字符串，max16M,224?1224?1 2^{24}-1224?1=16777215bytes
Text(16777216)	longtext	str	變長字符串，max32G,232?1232?1 2^{32}-1232?1=4294967295bytes
LargeBinary	blob	str	二進制文件，64K
LargeBinary(65536)	mediumblob	str	二進制，max16M
LargeBinary(16777216)	longblob	str	二進制，max32G
PickleType	blob	任何python對象	自動使用Pickle序列化，只有blob
Unicode	varchar	unicode	變長字符串
UnicodeText	text	unicode	變長字符串，64K
Boolean	tinyint	bool	布爾值
Date	date	datetime.date	日期
Time	time	date.time	時間
DateTime	datetime	datetime.datetime	日期和時間
Interval	datetime	datetime.timedelta	時間間隔
Enum	enum	str	一組字符串

屬性描述生效值

primary_key	主鍵	True
unique	鍵值唯一性	True
index	索引	True
nullable	空值	True
default	默認值	null

2.1.2?示例

#!/usr/bin/env python # -*- coding:utf-8 -*- import datetime from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index from sqlalchemy.orm import relationshipBase = declarative_base()###################### 單表示例 ######################### class Users(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)name = Column(String(32), index=True)age = Column(Integer, default=18)email = Column(String(32), unique=True)ctime = Column(DateTime, default=datetime.datetime.now)extra = Column(Text, nullable=True)__table_args__ = (UniqueConstraint('id', 'name', name='uix_id_name'), # 聯合唯一約束Index('ix_id_name', 'name', 'extra'), # 普通索引)class Hosts(Base):__tablename__ = 'hosts'id = Column(Integer, primary_key=True)name = Column(String(32), index=True)ctime = Column(DateTime, default=datetime.datetime.now)###################### 一對多示例 ######################### class Hobby(Base):__tablename__ = 'hobby'id = Column(Integer, primary_key=True)caption = Column(String(50), default='籃球')class Person(Base):__tablename__ = 'person'nid = Column(Integer, primary_key=True)name = Column(String(32), index=True, nullable=True)hobby_id = Column(Integer, ForeignKey("hobby.id"))# 與生成表結構無關，僅用于查詢方便hobby = relationship("Hobby", backref='pers') # 第一個字段表示關聯的表，backref 表示反向查詢的字段名稱###################### 多對多示例 ######################### class Server2Group(Base):__tablename__ = 'server2group'id = Column(Integer, primary_key=True, autoincrement=True)server_id = Column(Integer, ForeignKey('server.id'))group_id = Column(Integer, ForeignKey('group.id'))class Group(Base):__tablename__ = 'group'id = Column(Integer, primary_key=True)name = Column(String(64), unique=True, nullable=False)# 與生成表結構無關，僅用于查詢方便servers = relationship('Server', secondary='server2group', backref='groups') # secondary表示第三張表class Server(Base):__tablename__ = 'server'id = Column(Integer, primary_key=True, autoincrement=True)hostname = Column(String(64), unique=True, nullable=False)def init_db():"""根據類創建數據庫表"""engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",max_overflow=0, # 超過連接池大小外最多創建的連接pool_size=5, # 連接池大小pool_timeout=30, # 池中沒有線程最多等待的時間，否則報錯pool_recycle=-1 # 多久之后對線程池中的線程進行一次連接的回收（重置）)Base.metadata.create_all(engine)def drop_db():"""根據類刪除數據庫表"""engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",max_overflow=0, # 超過連接池大小外最多創建的連接pool_size=5, # 連接池大小pool_timeout=30, # 池中沒有線程最多等待的時間，否則報錯pool_recycle=-1 # 多久之后對線程池中的線程進行一次連接的回收（重置）)Base.metadata.drop_all(engine)if __name__ == '__main__':drop_db()init_db()

指定關聯列：hobby = relationship("Hobby", backref='pers',foreign_keys="Person.hobby_id")

2.2? 操作數據庫表

#!/usr/bin/env python # -*- coding:utf-8 -*- from?sqlalchemy.orm?import?sessionmaker from?sqlalchemy?import?create_engine from?models?import?Usersengine?=?create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5) Session?=?sessionmaker(bind=engine) session?=?Session() # 每次執行數據庫操作時，都需要創建一個session# ############# 執行ORM操作 ############# obj1?=?Users(name="alex1") session.add(obj1)session.commit() # 提交事務 session.close() # 關閉session

?????基于scoped_session實現線程安全

#!/usr/bin/env python # -*- coding:utf-8 -*- from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine from sqlalchemy.orm import scoped_session from models import Usersengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5) Session = sessionmaker(bind=engine) session = scoped_session(Session) """ # 線程安全，基于本地線程實現每個線程用同一個session # 特殊的：scoped_session中有原來方法的Session中的以下方法：public_methods = ('__contains__', '__iter__', 'add', 'add_all', 'begin', 'begin_nested','close', 'commit', 'connection', 'delete', 'execute', 'expire','expire_all', 'expunge', 'expunge_all', 'flush', 'get_bind','is_modified', 'bulk_save_objects', 'bulk_insert_mappings','bulk_update_mappings','merge', 'query', 'refresh', 'rollback','scalar' ) """# ############# 執行ORM操作 ############# obj1 = Users(name="alex1") session.add(obj1)session.commit() # 提交事務 session.close() # 關閉session

多線程執行示例

2.3? 基本增刪改查示例

#!/usr/bin/env python # -*- coding:utf-8 -*- import time import threadingfrom sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index from sqlalchemy.orm import sessionmaker, relationship from sqlalchemy import create_engine from sqlalchemy.sql import textfrom db import Users, Hostsengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5) Session = sessionmaker(bind=engine) session = Session()# ################ 添加 ################ # 單個插入 obj1 = Users(name="fenglepeng") session.add(obj1) session.flush() print(obj1.id) # 輸出新插入數據的主鍵，當只有一個主鍵時才能生效，如果設置兩個及以上會取不到插入的數據# 批量插入 session.add_all([Users(name="wupeiqi"),Users(name="alex"),Hosts(name="c1.com"), ]) session.commit() # 此時數據才插入到數據庫中# ################ 刪除 ################ session.query(Users).filter(Users.id > 2).delete() session.commit()# ################ 修改 ################ # 使用 update session.query(Users).filter(Users.id > 0).update({"name" : "099"}) session.query(Users).filter(Users.id > 0).update({Users.name: Users.name + "099"}, synchronize_session=False) # 字符串 session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, synchronize_session="evaluate") # 數字# 或者直接修改查詢到的對象的屬性 obj1 = session.query(Users).filter(Users.id > 0).first() obj1.name = 'fenglp' session.commit()# ################ 查詢 ################ r1 = session.query(Users).all() # 不加all，顯示的是執行的sql r2 = session.query(Users.name.label('xx'), Users.age).all() # label 別名 r3 = session.query(Users).filter(Users.name == "alex").all() # query表示查詢的字段，filter表示過濾的字段 r4 = session.query(Users).filter_by(name='alex').all() # 本表的字段 r5 = session.query(Users).filter_by(name='alex').first()# params 和Python中的format類似，用于替代前面的值 r6 = session.query(Users).filter(text("id<:value and name=:name")).params(value=224, name='fred').order_by(Users.id).all() r7 = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name='ed').all() # 嵌套查詢，子查詢session.close()

session 相關函數

db.session.add(obj) # 添加 db.session.flush(obj) # flush之后，可以取到obj的主鍵，但是當只有一個主鍵時才能生效，如果設置兩個及以上會取不到插 db.session.rollback() # 回滾 db.session.commit() ? # 提交事務，commit執行了，但是此時sql層面的commit并沒有執行 db.session.close()? ? # 關閉session

Model.query是db.session.query(Model)的快捷方式，它不可調用。?如果您沒有查詢模型，請繼續使用db.session.query(...)就像使用常規SQLAlchemy一樣。

# 下面兩種方式結果相同，第二種是第一種的簡化版 db.session.query(Users).all() # 這種query中可以加參數 Users.query.all() # 這種query中不能加參數

filter_by和filter的區別

filter_by表內部精確查詢

User.query.filter_by(id=4).first()

filter 全局查詢 id必須指明來源于那張表User，而且需要用等號，而不是賦值，建議使用filter

User.query.filter(User.id==4).first()

查詢操作

################# 條件查詢 ################ ret = session.query(Users).filter(Users.id = 1).all() # 等于 ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all() # between ret = session.query(Users).filter(Users.id.in_([1,3,4])).all() # in ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all() # 取反 ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all() # 子查詢################# 組合查詢 ################ ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all() # 組合查詢，默認是andfrom sqlalchemy import and_, or_ ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all() # _and里面的關系是and ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all() ret = session.query(Users).filter(or_(Users.id < 2,and_(Users.name == 'eric', Users.id > 3),Users.extra != "")).all()################# 通配符 % 表是多個，_ 表示一個 ################ ret = session.query(Users).filter(Users.name.like('e%')).all() ret = session.query(Users).filter(~Users.name.like('e%')).all()################# 限制 ################ ret = session.query(Users).filter(Users.name.like('e%')).limit(10).all() # offset ret = session.query(Users).filter(Users.name.like('e%')).limit(10).offset(100).all() # offset 和 limit ## 切片 ret = session.query(Users).filter(Users.name.like('e%')).slice(90,100).all() ret = session.query(Users).filter(Users.name.like('e%'))[90:100]################# 排序 ################ ret = session.query(Users).order_by(Users.name.desc()).all() ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()################# 去重 ################ ret = session.query(Users).order_by(Users.name.desc()).distinct().all()################# 分組 ################ from sqlalchemy.sql import func################# 分組聚合 ################ ret = session.query(Users).group_by(Users.extra).all() ret = session.query(func.max(Users.id),func.sum(Users.id),func.min(Users.id)).group_by(Users.name).all() ret = session.query(func.max(Users.id),func.sum(Users.id),func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()################# 連表 ################ # 已經設置好外鍵,它自動找到關聯的字段 ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all() ret = session.query(Person).join(Favor).all() # 默認是 inner join ret = session.query(Person).join(Favor, isouter=True).all() # isouter=True 表示左鏈接# 自己指定關聯關系 ret = session.query(Person).join(Favor, Person.id == Favor.nid).all() # 默認是 inner join ret = session.query(Person).join(Favor, Person.id == Favor.nid, isouter=True).all() # isouter=True 表示左鏈接,或者也可以 ret = session.query(Person).outerjoin(Favor, Person.id == Favor.nid).all() # 左連接# 返回值是一個列表，包含你查詢的兩個對象，如[<app.models.Favor object at 0x7f0a12692290>, <app.models.Person object at 0x7f0a12692290>]################# 組合 ################ q1 = session.query(Users.name).filter(Users.id > 2) q2 = session.query(Favor.caption).filter(Favor.nid < 2) ret = q1.union(q2).all()q1 = session.query(Users.name).filter(Users.id > 2) q2 = session.query(Favor.caption).filter(Favor.nid < 2) ret = q1.union_all(q2).all()

基于relationship操作ForeignKey

基于relationship操作m2m

執行原生SQL語句

################# 第一種 ################ import?time import?threading import?sqlalchemy from?sqlalchemy?import?create_engine from?sqlalchemy.engine.base?import?Engineengine?=?create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1?charset=utf8",max_overflow=0,??# 超過連接池大小外最多創建的連接pool_size=5,?? # 連接池大小pool_timeout=30,?# 池中沒有線程最多等待的時間，否則報錯pool_recycle=-1??# 多久之后對線程池中的線程進行一次連接的回收（重置） )def?task(arg):conn?=?engine.raw_connection()cursor?=?conn.cursor()cursor.execute("select * from t1")result?=?cursor.fetchall()cursor.close()conn.close()for?i?in?range(20):t?=?threading.Thread(target=task, args=(i,))t.start()################# 第二種 ################ #!/usr/bin/env python # -*- coding:utf-8 -*- import time import threading import sqlalchemy from sqlalchemy import create_engine from sqlalchemy.engine.base import Engineengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=0, pool_size=5)def task(arg):conn = engine.contextual_connect()with conn:cur = conn.execute("select * from t1")result = cur.fetchall()print(result)for i in range(20):t = threading.Thread(target=task, args=(i,))t.start()################# 第三種 ################ #!/usr/bin/env python # -*- coding:utf-8 -*- import time import threading import sqlalchemy from sqlalchemy import create_engine from sqlalchemy.engine.base import Engine from sqlalchemy.engine.result import ResultProxy engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=0, pool_size=5)def task(arg):cur = engine.execute("select * from t1")result = cur.fetchall()cur.close()print(result)for i in range(20):t = threading.Thread(target=task, args=(i,))t.start()################# 第四種 ################ #!/usr/bin/env python # -*- coding:utf-8 -*- import time import threadingfrom sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index from sqlalchemy.orm import sessionmaker, relationship from sqlalchemy import create_engine from sqlalchemy.sql import text from sqlalchemy.engine.result import ResultProxy from db import Users, Hostsengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5) Session = sessionmaker(bind=engine) session = Session()# 關聯子查詢 subqry = session.query(func.count(Server.id).label("sid")).filter(Server.id == Group.id).correlate(Group).as_scalar() result = session.query(Group.name, subqry) """ SELECT `group`.name AS group_name, (SELECT count(server.id) AS sid FROM server WHERE server.id = `group`.id) AS anon_1 FROM `group` """# 查詢 # cursor = session.execute('select * from users') # result = cursor.fetchall()# 添加 cursor = session.execute('insert into users(name) values(:value)',params={"value":'wupeiqi'}) session.commit() print(cursor.lastrowid)session.close()

注意: 查看連接?show status like 'Threads%';

從 datetime 字段中獲取當日數據

from sqlalchemy import funcclass History(db.Model):__tablename__ = 'historys'id = db.Column(db.Integer, primary_key=True)date = db.Column(db.Date)datetime = db.Column(db.Datetime)# firstDay：某年某月的第一天，datetime類型 # lastDay：某年某月的最后一天，datetime類型 historys = History.query.filter(History.date.between(firstDay, lastDay)).all() historys = History.query.filter(func.date(datetime).between(firstDay, lastDay)).all()# 格式化時間 func.date_format(History.datetime , "%Y-%m-%d %H:%i:%s") # 將時間戳轉化為時間 func.from_unixtime((Table.timestamp), "%Y-%m-%d %H:%i:%s") # 獲取12月份的所有數據 historys = History.query.filter(extract('month', History.date) == 12, extract('month', History.date) == 12).all()

動態增加條件

filter_by

filter_by用于查詢簡單的列名，不支持比較運算符。

filters = {’name': ‘fengyao', ‘age': 26} User.query.filter_by(**filters).first()

filter

比filter_by的功能更強大，支持比較運算符，支持or_、in_等語法。

filters = {User.name == ‘fengyao’,User.age > 25 } User.query.filter(*filters).first()

總結

以上是生活随笔為你收集整理的Flask 第三方组件之 SQLAlchemy的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇： Android使用Zxing库生成PDF
下一篇： qt release打包发布_几种解决Q