當(dāng)前位置：首頁(yè) > 运维知识 > 数据库 >内容正文

数据库

Flask 第三方组件之 SQLAlchemy

發(fā)布時(shí)間：2023/12/20 数据库 34 豆豆

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

一、介紹

SQLAlchemy是一個(gè)基于Python實(shí)現(xiàn)的ORM框架。該框架建立在 DB API之上，使用關(guān)系對(duì)象映射進(jìn)行數(shù)據(jù)庫(kù)操作，簡(jiǎn)言之便是：將類(lèi)和對(duì)象轉(zhuǎn)換成SQL，然后使用數(shù)據(jù)API執(zhí)行SQL并獲取執(zhí)行結(jié)果。

安裝：pip3 install sqlalchemy

組成部分：

Engine，框架的引擎
Connection Pooling ，數(shù)據(jù)庫(kù)連接池
Dialect，選擇連接數(shù)據(jù)庫(kù)的DB API種類(lèi)
Schema/Types，架構(gòu)和類(lèi)型
SQL Exprression Language，SQL表達(dá)式語(yǔ)言

SQLAlchemy本身無(wú)法操作數(shù)據(jù)庫(kù)，其必須以來(lái)pymsql等第三方插件，Dialect用于和數(shù)據(jù)API進(jìn)行交流，根據(jù)配置文件的不同調(diào)用不同的數(shù)據(jù)庫(kù)API，從而實(shí)現(xiàn)對(duì)數(shù)據(jù)庫(kù)的操作，如：

#################### 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??創(chuàng)建數(shù)據(jù)庫(kù)表

2.1.1 數(shù)據(jù)類(lèi)型和數(shù)據(jù)庫(kù)的對(duì)應(yīng)關(guān)系

類(lèi)型名MySQL類(lèi)型python類(lèi)型描述

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

屬性描述生效值

primary_key	主鍵	True
unique	鍵值唯一性	True
index	索引	True
nullable	空值	True
default	默認(rèn)值	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'), # 聯(lián)合唯一約束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)###################### 一對(duì)多示例 ######################### 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"))# 與生成表結(jié)構(gòu)無(wú)關(guān)，僅用于查詢(xún)方便hobby = relationship("Hobby", backref='pers') # 第一個(gè)字段表示關(guān)聯(lián)的表，backref 表示反向查詢(xún)的字段名稱(chēng)###################### 多對(duì)多示例 ######################### 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)# 與生成表結(jié)構(gòu)無(wú)關(guān)，僅用于查詢(xún)方便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():"""根據(jù)類(lèi)創(chuàng)建數(shù)據(jù)庫(kù)表"""engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",max_overflow=0, # 超過(guò)連接池大小外最多創(chuàng)建的連接pool_size=5, # 連接池大小pool_timeout=30, # 池中沒(méi)有線(xiàn)程最多等待的時(shí)間，否則報(bào)錯(cuò)pool_recycle=-1 # 多久之后對(duì)線(xiàn)程池中的線(xiàn)程進(jìn)行一次連接的回收（重置）)Base.metadata.create_all(engine)def drop_db():"""根據(jù)類(lèi)刪除數(shù)據(jù)庫(kù)表"""engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",max_overflow=0, # 超過(guò)連接池大小外最多創(chuàng)建的連接pool_size=5, # 連接池大小pool_timeout=30, # 池中沒(méi)有線(xiàn)程最多等待的時(shí)間，否則報(bào)錯(cuò)pool_recycle=-1 # 多久之后對(duì)線(xiàn)程池中的線(xiàn)程進(jìn)行一次連接的回收（重置）)Base.metadata.drop_all(engine)if __name__ == '__main__':drop_db()init_db()

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

2.2? 操作數(shù)據(jù)庫(kù)表

#!/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() # 每次執(zhí)行數(shù)據(jù)庫(kù)操作時(shí)，都需要?jiǎng)?chuàng)建一個(gè)session# ############# 執(zhí)行ORM操作 ############# obj1?=?Users(name="alex1") session.add(obj1)session.commit() # 提交事務(wù) session.close() # 關(guān)閉session

?????基于scoped_session實(shí)現(xiàn)線(xiàn)程安全

#!/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) """ # 線(xiàn)程安全，基于本地線(xiàn)程實(shí)現(xiàn)每個(gè)線(xiàn)程用同一個(gè)session # 特殊的：scoped_session中有原來(lái)方法的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' ) """# ############# 執(zhí)行ORM操作 ############# obj1 = Users(name="alex1") session.add(obj1)session.commit() # 提交事務(wù) session.close() # 關(guān)閉session

多線(xiàn)程執(zhí)行示例

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()# ################ 添加 ################ # 單個(gè)插入 obj1 = Users(name="fenglepeng") session.add(obj1) session.flush() print(obj1.id) # 輸出新插入數(shù)據(jù)的主鍵，當(dāng)只有一個(gè)主鍵時(shí)才能生效，如果設(shè)置兩個(gè)及以上會(huì)取不到插入的數(shù)據(jù)# 批量插入 session.add_all([Users(name="wupeiqi"),Users(name="alex"),Hosts(name="c1.com"), ]) session.commit() # 此時(shí)數(shù)據(jù)才插入到數(shù)據(jù)庫(kù)中# ################ 刪除 ################ 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") # 數(shù)字# 或者直接修改查詢(xún)到的對(duì)象的屬性 obj1 = session.query(Users).filter(Users.id > 0).first() obj1.name = 'fenglp' session.commit()# ################ 查詢(xún) ################ r1 = session.query(Users).all() # 不加all，顯示的是執(zhí)行的sql r2 = session.query(Users.name.label('xx'), Users.age).all() # label 別名 r3 = session.query(Users).filter(Users.name == "alex").all() # query表示查詢(xún)的字段，filter表示過(guò)濾的字段 r4 = session.query(Users).filter_by(name='alex').all() # 本表的字段 r5 = session.query(Users).filter_by(name='alex').first()# params 和Python中的format類(lèi)似，用于替代前面的值 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() # 嵌套查詢(xún)，子查詢(xún)session.close()

session 相關(guān)函數(shù)

db.session.add(obj) # 添加 db.session.flush(obj) # flush之后，可以取到obj的主鍵，但是當(dāng)只有一個(gè)主鍵時(shí)才能生效，如果設(shè)置兩個(gè)及以上會(huì)取不到插 db.session.rollback() # 回滾 db.session.commit() ? # 提交事務(wù)，commit執(zhí)行了，但是此時(shí)sql層面的commit并沒(méi)有執(zhí)行 db.session.close()? ? # 關(guān)閉session

Model.query是db.session.query(Model)的快捷方式，它不可調(diào)用。?如果您沒(méi)有查詢(xún)模型，請(qǐng)繼續(xù)使用db.session.query(...)就像使用常規(guī)SQLAlchemy一樣。

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

filter_by和filter的區(qū)別

filter_by表內(nèi)部精確查詢(xún)

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

filter 全局查詢(xún) id必須指明來(lái)源于那張表User，而且需要用等號(hào)，而不是賦值，建議使用filter

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

查詢(xún)操作

################# 條件查詢(xún) ################ 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() # 子查詢(xún)################# 組合查詢(xún) ################ ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all() # 組合查詢(xún)，默認(rèn)是andfrom sqlalchemy import and_, or_ ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all() # _and里面的關(guān)系是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()################# 通配符 % 表是多個(gè)，_ 表示一個(gè) ################ 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()################# 連表 ################ # 已經(jīng)設(shè)置好外鍵,它自動(dòng)找到關(guān)聯(lián)的字段 ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all() ret = session.query(Person).join(Favor).all() # 默認(rèn)是 inner join ret = session.query(Person).join(Favor, isouter=True).all() # isouter=True 表示左鏈接# 自己指定關(guān)聯(lián)關(guān)系 ret = session.query(Person).join(Favor, Person.id == Favor.nid).all() # 默認(rèn)是 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() # 左連接# 返回值是一個(gè)列表，包含你查詢(xún)的兩個(gè)對(duì)象，如[<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

執(zhí)行原生SQL語(yǔ)句

################# 第一種 ################ 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,??# 超過(guò)連接池大小外最多創(chuàng)建的連接pool_size=5,?? # 連接池大小pool_timeout=30,?# 池中沒(méi)有線(xiàn)程最多等待的時(shí)間，否則報(bào)錯(cuò)pool_recycle=-1??# 多久之后對(duì)線(xiàn)程池中的線(xiàn)程進(jìn)行一次連接的回收（重置） )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()# 關(guān)聯(lián)子查詢(xún) 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` """# 查詢(xún) # 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 字段中獲取當(dāng)日數(shù)據(jù)

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類(lèi)型 # lastDay：某年某月的最后一天，datetime類(lèi)型 historys = History.query.filter(History.date.between(firstDay, lastDay)).all() historys = History.query.filter(func.date(datetime).between(firstDay, lastDay)).all()# 格式化時(shí)間 func.date_format(History.datetime , "%Y-%m-%d %H:%i:%s") # 將時(shí)間戳轉(zhuǎn)化為時(shí)間 func.from_unixtime((Table.timestamp), "%Y-%m-%d %H:%i:%s") # 獲取12月份的所有數(shù)據(jù) historys = History.query.filter(extract('month', History.date) == 12, extract('month', History.date) == 12).all()

動(dòng)態(tài)增加條件

filter_by

filter_by用于查詢(xún)簡(jiǎn)單的列名，不支持比較運(yùn)算符。

filters = {’name': ‘fengyao', ‘a(chǎn)ge': 26} User.query.filter_by(**filters).first()

filter

比f(wàn)ilter_by的功能更強(qiáng)大，支持比較運(yùn)算符，支持or_、in_等語(yǔ)法。

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

總結(jié)

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

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。

上一篇： Android使用Zxing库生成PDF
下一篇： linux cmake编译源码,linu