SQLAlchemy是Python編程語言下的一款ORM框架,該框架創建在數據庫API之上,使用關係對象映射進行數據庫操做,簡言之即是:將對象轉換成SQL,而後使用數據API執行SQL並獲取執行結果。html
組成部分:python
SQLAlchemy自己沒法操做數據庫,其必須以來pymsql等第三方插件,Dialect用於和數據API進行交流,根據配置文件的不一樣調用不一樣的數據庫API,從而實現對數據庫的操做,如:mysql
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
sql
pip install SQLAlchemy
create_engine 方法進行數據庫鏈接,返回一個 db 對象。參數echo = True能夠在控制檯打印sql語句
經過這個engine對象能夠直接execute 進行查詢,例如 engine.execute("SELECT * FROM user") 也能夠經過 engine 獲取鏈接在查詢,例如 conn = engine.connect() 經過 conn.execute()方法進行查詢。二者有什麼差異呢?數據庫
主要差異在因而否使用transaction模式, 若是不涉及transaction, 兩種方法效果是同樣的. 官網推薦使用後者。django
使用engine的execute執行sql:編程
from sqlalchemy import create_engine engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/s8?charset=utf8") cur = engine.execute('select * from users') result = cur.fetchall() print(result)
使用engine.connect()執行sql語句:安全
from sqlalchemy import create_engine engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/s8?charset=utf8") conn = engine.connect() cur = conn.execute('select * from users') result = cur.fetchall() print(result)
使用鏈接池session
from sqlalchemy import create_engine from sqlalchemy.engine.base import Engine engine = create_engine( "mysql+pymysql://root:123456@127.0.0.1:3306/s8?charset=utf8", max_overflow=0, # 超過鏈接池大小外最多建立的鏈接 pool_size=5, # 鏈接池大小 pool_timeout=30, # 池中沒有線程最多等待的時間,不然報錯 pool_recycle=-1 # 多久以後對線程池中的線程進行一次鏈接的回收(重置) ) conn = engine.raw_connection() cursor = conn.cursor() cursor.execute("select * from users") result = cursor.fetchall() print(result) cursor.close() conn.close()
from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index # 建立鏈接 engine = create_engine("mysql+pymysql://root:root@192.168.56.11/oldboydb",encoding='utf-8',echo=True) #生成ORM基類 Base=declarative_base() class User(Base): __tablename__ = 'user' #表名 id = Column(Integer,primary_key=True) #字段,整形,主鍵 column是導入的 name = Column(String(32)) password = Column(String(64)) Base.metadata.create_all(engine) #在engine鏈接的數據庫裏建立表結構
經常使用字段架構
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 中表現爲 Decimal . LargeBinary (length=None) # 字節數據. 根據數據庫實現, 在實例化時可能須要指定大小. PickleType # Python 對象的序列化類型. String (length=None, collation=None, ...) # 字符串類型, Python 中表現爲 Unicode , 數據庫表現爲 VARCHAR , 一般都須要指定長度. Unicode # 相似與字符串類型, 在某些數據庫實現下, 會明確表示支持非 ASCII 字符. 同時輸入輸出也強制是 Unicode 類型. Text # 長文本類型, Python 表現爲 Unicode , 數據庫表現爲 TEXT .
Column指定的一些字段參數
default # 默認值,時間字段的默認值datetime.datetime.now不能加(),不然會執行生成固定值 primary_key=True # 設置爲主鍵 autoincrement=True # 主鍵的自增 index=True # 做爲索引 nullable=True # 是否能夠爲空
表的參數
# 在類下定義__table_args__ __table_args__ = ( # UniqueConstraint('id', 'name', name='uix_id_name'),聯合惟一 # Index('ix_id_name', 'name', 'extra'),聯合索引 # 'mysql_engine': 'InnoDB', # 'mysql_charset': 'utf8' )
外鍵設置
# sqlaichemy不像django擁有外鍵字段,設置外鍵須要導入ForeignKey指定哪張表的那個字段,做爲Column的第二個參數 from sqlalchemy import Column, Integer, ForeignKey # 字段 hobby_id = Column(Integer, ForeignKey("hobby.id"))
另外若是是多對多的表,他也沒法建立第三張表,須要手動建立第三張表關聯兩張表
一張簡單的表
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 relationship Base = 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'), ) # 生成這張表 engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/s8?charset=utf8") Base.metadata.create_all(engine) # 將繼承了Base類的表在engine鏈接的數據庫中生成 # 刪除表 engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/s8?charset=utf8") Base.metadata.drop_all(engine) # 將繼承了Base類的表在engine鏈接的數據庫中刪除
一對多關係的設置
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 relationship Base = declarative_base() class Classes(Base): __tablename__ = 'classes' id = Column(Integer, primary_key=True) name = Column(String(32), index=True) class Student(Base): __tablename__ = 'student' id = Column(Integer, primary_key=True) name = Column(String(32), index=True) age = Column(Integer, default=18) classes = Column(Integer,ForeignKey('classes.id')) # 與生成表結構無關,僅用於查詢方便,指定了關聯的表,和反向查詢的名字 # 他會根據這個類裏與指定表關聯的字段去查找 hobby = relationship("Classes", backref='stu')
多對多關係的表設計
class Classes(Base): __tablename__ = 'classes' id = Column(Integer, primary_key=True) name = Column(String(32), index=True) class Teacher(Base): __tablename__ = 'teacher' id = Column(Integer, primary_key=True) name = Column(String(32), index=True) age = Column(Integer, default=19) # 與生成表結構無關,僅用於查詢方便,指定了關聯的表,中間表的表名和反向查詢的名字 servers = relationship('Classes', secondary='server2group', backref='classes') class Teacher2Class(Base): __tablename__ = 'server2group' id = Column(Integer, primary_key=True) classes_id = Column(Integer,ForeignKey('classes.id')) teacher_id = Column(Integer,ForeignKey('teacher.id')) __table_args__ = ( UniqueConstraint('classes_id', 'teacher_id', name='tea_cls'), # Index('ix_id_name', 'name', 'extra'), )
ORM經過Session與數據庫創建鏈接的。當應用第一次載入時,咱們定義一個Session類(聲明create_engine()的同時),這個Session類爲新的Session對象提供工廠服務。
from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine engine =create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/s8day128db?charset=utf8") Session = sessionmaker(bind=engine)
建立鏈接的方式
# 方式一 # 這個定製的Session類會建立綁定到數據庫的Session對象。若是須要和數據庫創建鏈接,只須要實例化一個Session: session = Session() # 此處只可以在視圖函數內執行 # 執行數據庫操做 # 修改操做進行session.commit() session.close() # 方式二:支持線程安全,爲每一個線程建立一個session session = scoped_session(Session) # 執行數據庫操做 # 修改操做進行session.commit() session.remove()
方式二的源碼:
class scoped_session(object): session_factory = None def __init__(self, session_factory, scopefunc=None): self.session_factory = session_factory # 原來的Session if scopefunc: self.registry = ScopedRegistry(session_factory, scopefunc) else: self.registry = ThreadLocalRegistry(session_factory) def __call__(self, **kw): if kw: if self.registry.has(): raise sa_exc.InvalidRequestError( "Scoped session is already present; " "no new arguments may be specified.") else: sess = self.session_factory(**kw) self.registry.set(sess) return sess else: return self.registry() ... class ThreadLocalRegistry(ScopedRegistry): def __init__(self, createfunc): self.createfunc = createfunc # 原來的Session self.registry = threading.local() def __call__(self): try: return self.registry.value except AttributeError: val = self.registry.value = self.createfunc() return val def instrument(name): def do(self, *args, **kwargs): return getattr(self.registry(), name)(*args, **kwargs) return do # 這裏返回的是函數,至關於self.query = do,self.query()至關於do() for meth in Session.public_methods: # meth就是原Session對象的屬性 setattr(scoped_session, meth, instrument(meth))
添加一條
clsobj = models.Classes(name='全棧1期') # 實例化類 session.add(clsobj) # 經過session將clsobj添加進數據庫中 # 上面並不須要指定庫,由於clsobj是Classes的實例,他們存在着對應關係 session.commit() # 提交
添加多條
clsobj = models.Classes(name='全棧2期') stuobj = models.Student(name="李淳罡",classes=1) session.add_all([clsobj,stuobj]) # 也正是由於對應關係的存在,咱們能夠將不一樣類的實例一塊兒提交 session.commit()
r1 = session.query(models.Classes).all() print(r1) # [<models.Classes object at 0x0000020B3F849240>, <models.Classes object at 0x0000020B3F8492E8>] r2 = session.query(models.Classes.name.label('xx'), models.Classes.id).all() print(r2,type(r2[0])) # [('全棧1期', 1), ('全棧2期', 2)] <class 'sqlalchemy.util._collections.result'> # 看似元組其實並非,也能夠經過.字段的方式得到值,.label('xx')至關於爲這個字段從新齊了名字,至關於sql中的as r3 = session.query(models.Classes).filter(models.Classes.name == "全棧1期").all() print(r3) # filter(表達式) [<models.Classes object at 0x0000020B3F849240>] r4 = session.query(models.Classes).filter_by(name='alex').all() print(r4) # filter_by(字段=值) [] r5 = session.query(models.Classes).filter_by(name='alex').first() print(r5) # None
session.query(models.Classes).filter(models.Classes.id>2).delete() # 此處應注意是什麼調用的.delete(),all()方法返回的是個列表 session.commit()
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") # updata方法傳遞字典的鍵能夠是字符串類型的字段名,也能夠是表下的字段, # 若是是對原數據進行修改還要指定synchronize_session
# 條件 ret = session.query(Users).filter_by(name='alex').all() # 多條件,隔開,and關係 ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all() # between在a,b之間 ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all() # in_(),接收一個列表 ret = session.query(Users).filter(Users.id.in_([1,3,4])).all() # ~取反 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_ # 查詢條件的關係 # and關係 ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all() # or關係 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() # 通配符,模糊匹配like() ret = session.query(Users).filter(Users.name.like('e%')).all() ret = session.query(Users).filter(~Users.name.like('e%')).all() # 限制 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() # souter=True左鏈接,他們的鏈接不用設置關聯字段,由於他們之間存在外鍵關係 ret = session.query(Person).join(Favor, isouter=True).all() # 無外鍵關聯則須要本身設置關聯字段,query什麼就能查到什麼,不能.別的屬性,通常用做關聯查詢 obj = session.query(models.Student).join(models.Userinfo,models.Student.id==models.Userinfo.user_id).first() # 組合,將具備相同字段數量的查詢結果聯合成一個結果 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() # 不去重 # 原生查詢,查詢結果必須在query中 result = session.query(models.Classes).from_statement(text("SELECT * FROM student where age=:age")).params(age=18).first() obj = session.query("name","age").from_statement(text('SELECT name,age from teacher')).all()
跨表查詢實例
# 找到全部學生,打印學生信息(包含班級名稱) # 子查詢 objs = session.query(models.Student).all() for obj in objs: cls_obj = session.query(models.Classes).filter(models.Classes.id==obj.classes).first() print(obj.id,obj.name,obj.classes,cls_obj.name) # 連表,已有外鍵關聯 objs = session.query(models.Student.id,models.Student.name,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.name,item.classes,item.cls.name)
另外relationship()還能夠用來添加數據
# 一對多示例 # 向Student增長一條記錄,順便向Classes增長一條記錄 session.add(models.Student(name='小韓',cls=models.Classes(name='全棧8期'))) # 向學生表增長一條記錄 clsobj = session.query(models.Classes).filter(models.Classes.name == "全棧8期").first() session.add(models.Student(name='崔絲塔娜',cls=clsobj)) # 反向操做 # 建立班級同時建立學生 # 由於是一對多的關係,因此stu應該是一個集合 session.add(models.Classes(name='全棧3期',stu=[models.Student(name='奧利安娜'),models.Student(name='莫甘娜')])) # 多對多 # 建立講師關聯班級,建立班級 obj = models.Teacher(name='奧菲娜') obj.servers = [models.Classes(name='全棧4期'),models.Classes(name='全棧5期')] session.add(obj) # 建立講師不建立班級 clsobj = session.query(models.Classes).filter(models.Classes.name == "全棧8期").first() session.add(models.Teacher(name='露露',servers=[clsobj])) # 反向添加 session.add(models.Classes(name='全棧9期',classes=[models.Teacher(name='奧瑞利亞'),models.Teacher(name='索拉卡')]))