SQLALchemy
SQLALchemy
SQLAlchemy是一個基於Python實現的ORM框架。該框架創建在 DB API之上，使用關係對象映射進行數據庫操做 ，也就是將類和對象轉換成SQL，而後使用數據API執行SQL並獲取執行結果。html
安裝：pip3 install sqlalchemypython
組成部分：mysql
- Engine，框架的引擎
- Connection Pooling ，數據庫鏈接池
- Dialect，選擇鏈接數據庫的DB API種類
- Schema/Types，架構和類型
- SQL Exprression Language，SQL表達式語言
SQLAlchemy自己沒法操做數據庫，其必須以來pymsql等第三方插件，Dialect用於和數據API進行交流，根據配置文件的不一樣調用不一樣的數據庫API，從而實現對數據庫的操做，如：sql
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
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
|
問題：SQLALchemy，ORM框架
一、什麼是orm框架
orm：對象關係映射
類 --- 表
對象 --- 一條記錄
當有了對應關係以後，不須要再編寫sql語句，直接操做，類，對象
二、sql vs orm
sql查詢速度快，開發效率低
orm開發效率高，查詢速度較低
三、概念理解
-db first 根據數據庫的表生成類
-code first 根據類生成數據庫的表
四、orm是怎麼實現的
根據對象和類經過字符串格式化轉化成sql語句
DDD中：unit of work
使用原生sql語句
import time import threading import sqlalchemy from sqlalchemy import create_engine from sqlalchemy.engine.base import Engine engine = 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 Engine engine = 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()
使用orm語句
建立數據庫表
from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, DateTime,ForeignKey,UniqueConstraint from sqlalchemy import create_engine from sqlalchemy.orm import relationship import datetime Base = declarative_base() # 建立單表 class Classes(Base): __tablename__ = 'classes' id = Column(Integer, primary_key=True,autoincrement=True) name = Column(String(32)) class Student(Base): __tablename__="student" id =Column(Integer,primary_key=True,autoincrement=True) user=Column(String(32)) pwd=Column(String(32)) ctime=Column(DateTime,default=datetime.datetime.now) #不要加() class_id=Column(Integer,ForeignKey("classes.id")) #不會生成字段，用於鏈表查詢，backref用於反向鏈表查表 cls = relationship("Classes", backref='stus') class Hobby(Base): __tablename__="hobby" id=Column(Integer,primary_key=True,autoincrement=True) caption=Column(String(32),default="大球") # 多對多關係 須要自檢建立第三張表 class StudentToHobby(Base): __tablename__="studenttohobby" id = Column(Integer, primary_key=True, autoincrement=True) student_id=Column(Integer,ForeignKey('student.id')) hobby_id=Column(Integer,ForeignKey('hobby.id')) # 聯合惟一 __table_args__=( UniqueConstraint("student_id","hobby_id",name="uix_student_id_hobby_id"), ) def init_db(): # 數據庫鏈接相關 engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/s8sql?charset=utf8") # 建立表 Base.metadata.create_all(engine) def drop_db(): engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/s8sql?charset=utf8") # 刪除表 Base.metadata.drop_all(engine) if __name__ == '__main__': init_db() # drop_db()
增
import models from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine # 鏈接數據庫 engine=create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/s8sql?charset=utf8") Session=sessionmaker(bind=engine) # 每次執行數據庫操做時，都須要建立一個session session=Session() # 添加單條數據 # obj=models.Classes(name="全棧8期") # session.add(obj) # 添加多條數據 # obj_all=[ # models.Classes(name="全棧9期"), # models.Classes(name="全棧10期"), # models.Classes(name="全棧11期") # ] # session.add_all(obj_all) # 添加數據（含有外鍵） obj=models.Student(user="小花",pwd="123",class_id=2) session.add(obj) # 提交事務 session.commit() # 關閉session session.close()
查
import models from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine,text # 鏈接數據庫 engine=create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/s8sql?charset=utf8") Session=sessionmaker(bind=engine) # 每次執行數據庫操做時，都須要建立一個session session=Session() # 查所有數據 # class_list=session.query(models.Classes).all() # for classes in class_list: # print(classes.id,classes.name) #別名查詢 # class_list=session.query(models.Classes.id,models.Classes.name.label("xx")).all() # for classes in class_list: # print(classes.id,classes.xx) #條件查詢 filter(條件) filter_by() # obj=session.query(models.Classes).filter(models.Classes.name=="全棧9期").all() # for i in obj: # print(i.name) # obj1=session.query(models.Classes).filter(models.Classes.name=="全棧9期").first() # print(obj1.name) # obj2=session.query(models.Classes).filter_by(name="全棧10期").first() # print(obj2.name) #子查詢 # result = session.query(models.Classes).from_statement(text("SELECT * FROM classes where name=:name")).params(name='全棧11期').all() # print(result) # ret = session.query(models.Classes).filter(models.Classes.id.in_(session.query(models.Classes.id).filter_by(name='全棧8期'))).all() # print(ret) # 顯示全部學生信息（含班級） # 一、屢次查詢 # obj=session.query(models.Student).all() # for obj in objs: # cls_obj = session.query(models.Classes).filter(models.Classes.id==obj.class_id).first() # print(obj.id,obj.username,obj.class_id,cls_obj.name) # 二、連表查詢 isouter=True mysql中leftjoin查詢 # objs = session.query(models.Student.id,models.Student.user,models.Classes.name).join(models.Classes,isouter=True).all() # print(objs) # 三、使用relationship查詢 # objs = session.query(models.Student).all() # for item in objs: # print(item.id,item.user,item.class_id,item.cls.name) #四、使用relationship反向查詢 全棧9期全部的學生 # obj = session.query(models.Classes).filter(models.Classes.name=='全棧9期').first() # student_list = obj.stus # for item in student_list: # print(item.id,item.user) # 提交事務 session.commit() # 關閉session session.close()
改
import models from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine # 鏈接數據庫 engine=create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/s8sql?charset=utf8") Session=sessionmaker(bind=engine) # 每次執行數據庫操做時，都須要建立一個session session=Session() # 改 # session.query(models.Classes).filter(models.Classes.id > 1).update({"name" : "099"}) # synchronize_session=False 字符串拼接 # session.query(models.Classes).filter(models.Classes.id > 0).update({models.Classes.name: models.Classes.name + "099"}, synchronize_session=False) # synchronize_session="evaluate" 數字計算 session.query(models.Classes).filter(models.Classes.id > 0).update({"age": models.Classes.age + 1}, synchronize_session="evaluate") # 提交事務 session.commit() # 關閉session session.close()
刪
import models from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine # 鏈接數據庫 engine=create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/s8sql?charset=utf8") Session=sessionmaker(bind=engine) # 每次執行數據庫操做時，都須要建立一個session session=Session() # 刪 session.query(models.Classes).filter(models.Classes.id > 2).delete() # 提交事務 session.commit() # 關閉session session.close()
補充操做
# 條件 ret = session.query(Users).filter_by(name='alex').all() ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').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() #id=1or3or 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() from sqlalchemy import and_, or_ ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all() 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() # 限制 （mysql中的limit） ret = session.query(Users)[1:2] # 排序 ret = session.query(Users).order_by(Users.name.desc()).all() ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).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() #left join 以左表爲基準 # 組合 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
- relationship做用？增長數據和查詢 一對多： class UserType(Base): __tablename__ = 'usertype' 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("usertype.id")) # 與生成表結構無關，僅用於查詢方便 hobby = relationship("UserType", backref='pers') 數據添加： session.add(UserType(caption='超級管理員')) session.commit() session.add(Person(name='翔龍',hobby_id=1)) session.commit() user表添加數據，同時也給UserType表添加數據 session.add(Person(name='小韓',hobby=UserType(cation='VVIP'))) session.commit() 數據查詢： 正向操做： obj = session.query(Person).filter(Person.nid==2).first() obj.hobby.caption 反向操做： obj = session.query(UserType).filter(UserType.id==1).first() obj.pers 多對多： class User2Hobby(Base): __tablename__ = 'user2hobby' id = Column(Integer, primary_key=True, autoincrement=True) hobby_id = Column(Integer, ForeignKey('hobby.id')) user_id = Column(Integer, ForeignKey('user.id')) class Hobby(Base): __tablename__ = 'hobby' id = Column(Integer, primary_key=True) title = Column(String(64), unique=True, nullable=False) # 與生成表結構無關，僅用於查詢方便 users = relationship('User', secondary='user2hobby', backref='hbs') class User(Base): __tablename__ = 'user' id = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(64), unique=True, nullable=False) 添加數據： 添加單條： session.add(Hobby(title='籃球')) session.commit() session.add(User(name='梅凱')) session.commit() session.add(User2Hobby(hobby_id=1,user_id=1)) session.commit() 添加多條 正向： obj = Hobby(title='籃球') obj.servers = [User(name='王巖'),User(name='曉梅')] session.add(obj) session.commit() 反向： obj = User(title='俊傑') obj.hbs = [Hobby(title='翔龍'),Hobby(title='興隆')] session.add(obj) session.commit() 查詢： 反向： obj = session.query(User).filter(User.id==2).first() obj.hbs 正向: obj = session.query(Hobby).filter(Hobby.id==2).first() obj.users
建立session的兩種方式
#!/usr/bin/env python # -*- coding:utf-8 -*- import time import threading from 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 db import Users engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5) Session = sessionmaker(bind=engine) def task(arg): session = Session() obj1 = Users(name="alex1") session.add(obj1) session.commit() for i in range(10): t = threading.Thread(target=task, args=(i,)) t.start()
#!/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 Users engine = 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中的一下方法： 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' ) """ session = scoped_session(Session) # ############# 執行ORM操做 ############# obj1 = Users(name="alex1") session.add(obj1) # 提交事務 session.commit() # 關閉session session.close()
flask-session默認也是使用的第二種方式：scoped_session數據庫
其它
- SQL a. select * from A where id in (select id from B) #不能使用*必須與id類型匹配 b. #1表示常量，每條數據都添加1 select id, name, 1 from A select id, name, 1, (select max(id) from B) as b from A select id, name, 1, (select max(id) from B where B.xid=A.id) as b from A +----+---------------+ | id | name | +----+---------------+ | 1 | 全棧1期099 | | 2 | 全棧2期099 | +----+---------------+ +----+---------+ | id | caption | xid +----+---------+ | 1 | 籃球 | 1 | 2 | 球 | 1 +----+---------+ 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` """ # 也可使用原生SQL """ # 查詢 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) """
相關文章
- 1. SQLAlchemy和Flask-SQLAlchemy
- 2. flask-SQLAlchemy django-SQLAlchemy orm
- 3. SQLAlchemy
- 4. sqlalchemy
- 5. SQLALCHEMY
- 6. SQLalchemy
- 7. SqlAlchemy
- 更多相關文章...
- • Docker Compose - Docker教程
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章