SQLAlchemy

時間  2019-11-17
標籤 sqlalchemy 简体版
原文   原文鏈接

 

一 介紹html

 

SQLAlchemy是Python編程語言下的一款ORM框架，該框架創建在數據庫API之上，使用關係對象映射進行數據庫操做，簡言之即是：將對象轉換成SQL，而後使用數據API執行SQL並獲取執行結果。python

1. 安裝：mysql

pip3 install sqlalchemy

2. 框架與流程sql

 

組成部分：數據庫

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

流程：編程

#一、使用者經過ORM對象提交命令
#二、將命令交給SQLAlchemy Core（Schema/Types  SQL Expression Language）轉換成SQL
#三、使用 Engine/ConnectionPooling/Dialect 進行數據庫操做
#3.一、匹配使用者事先配置好的egine
#3.二、egine從鏈接池中取出一個連接
#3.三、基於該連接經過Dialect調用DB API，將SQL轉交給它去執行
View Code

 

上述流程分析，能夠大體分爲兩個階段

#第一個階段（流程1-2）：將SQLAlchemy的對象換成可執行的sql語句

#第二個階段（流程3）：將sql語句交給數據庫執行
View Code

 

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

　　

二 使用session

 

執行原生SQL語句多線程

 

from sqlalchemy import create_engine

#1 準備
# 須要事先安裝好pymysql
# 須要事先建立好數據庫:create database db1 charset utf8;

#2 建立引擎
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  # 多久以後對線程池中的線程進行一次鏈接的回收（重置）
)

#3 執行sql
# egine.execute('create table if not EXISTS t1(id int PRIMARY KEY auto_increment,name char(32));')

# cur=egine.execute('insert into t1 values(%s,%s);',[(1,"zhou"),(2,"jun"),(3,"egon3")]) #按位置傳值

# cur=egine.execute('insert into t1 values(%(id)s,%(name)s);',name='hao',id=4) #按關鍵字傳值

#4 新插入行的自增id
# print(cur.lastrowid)

#5 查詢
cur=egine.execute('select * from t1')

cur.fetchone() #獲取一行
cur.fetchmany(2) #獲取多行
cur.fetchall() #獲取全部行

#############
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()


############
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()

注意: 查看鏈接 show status like 'Threads%';　　架構

 

 

建立表

 

#!/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

Base = declarative_base()


class Users(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False) #nullable爲False是不容許爲空
    # 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', 'email'),
           'mysql_engine': 'InnoDB', #指定存儲引擎
           'mysql_charset': 'utf8'  #指定表的字符編碼

    )


def init_db():
    """
    根據類建立數據庫表
    :return: 
    """
    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():
    """
    根據類刪除數據庫表
    :return: 
    """
    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()
建立單表 
#!/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 relationship

Base = declarative_base()


# ##################### 一對多示例 #########################
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')


# ##################### 多對多示例 #########################

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')


class Server(Base):
    __tablename__ = 'server'

    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String(64), unique=True, nullable=False)
建立多個表幷包含Fk、M2M關係

 

執行數據庫操做

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
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
session = Session()
  
# ############# 執行ORM操做 #############
obj1 = Users(name="alex1")
session.add(obj1)
  
# 提交事務
session.commit()
# 關閉session
session.close()

 

基本的增刪改查

#!/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 sqlalchemy.sql import text

from db import Users, Hosts

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/demo1", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)

session = Session()

# ################ 添加 ################
"""
obj1 = Users(name="wupeiqi")
session.add(obj1)

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()
"""
# ################ 修改 ################
"""
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) #非運算操做指定synchronize_session=False
session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, synchronize_session="evaluate")# 運算操做時指定參數爲synchronize_session="evaluate"
session.commit()
"""
# ################ 查詢 ################
"""
r1 = session.query(Users).all() # 查全部
r2 = session.query(Users.name.label('xx'), Users.age).all() #查指定字段
r3 = session.query(Users).filter(Users.name == "alex").all() # 值過濾
r4 = session.query(Users).filter_by(name='alex').all() # 屬性過濾
r5 = session.query(Users).filter_by(name='alex').first() # 取結果的第一個
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()

#!/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()
基於scoped_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()
多線程執行示例

 

 

經常使用操做

#　條件
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()

# in 與 notin
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()

# and 與 or
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() # e開頭對全部
ret = session.query(Users).filter(~Users.name.like('e%')).all() # ～取反

# 限制
ret = session.query(Users)[1:2:1]  #對結果切片

# 排序
# desc降序，asc升序
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()

ret = session.query(Person).join(Favor, isouter=True).all()


# 組合  

# 把兩次查詢的結果組合到一個列表
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反查

#!/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 sqlalchemy.sql import text
from sqlalchemy.engine.result import ResultProxy
from db import Users, Hosts, Hobby, Person

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()


# 使用relationship正向查詢
"""
v = session.query(Person).first()
print(v.name)
print(v.hobby.caption) # 結果是一個對象
"""

# 使用relationship反向查詢
"""
v = session.query(Hobby).first()
print(v.caption)
print(v.pers) # 結果是一個列表，列表內是關聯的全部對象
"""

session.close()

　　

多對多查詢及relationship反查

#!/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 sqlalchemy.sql import text
from sqlalchemy.engine.result import ResultProxy
from db import Users, Hosts, Hobby, Person, Group, Server, Server2Group

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()



# 使用relationship正向查詢
"""
v = session.query(Group).first()
print(v.name)
print(v.servers) # 獲得一個列表，裏面是全部關聯的對象
"""

# 使用relationship反向查詢
"""
v = session.query(Server).first()
print(v.hostname)
print(v.groups)  # 獲得一個列表，裏面是全部關聯的對象
"""


session.close()

 

子查詢

#示例:查出id大於2的員工,當作子查詢的表使用

#原生SQL:
# select * from (select * from emp where id > 2);

#ORM:
res=session.query(
    session.query(Emp).filter(Emp.id > 8).subquery()
).all()
形式一:子查詢當作一張表來用,調用subquery() 
#示例:#查出銷售部門的員工姓名

#原生SQL:
# select ename from emp where dep_id in (select id from dep where dname='銷售');

#ORM:
res=session.query(Emp.ename).filter(Emp.dep_id.in_(
    session.query(Dep.id).filter_by(dname='銷售'), #傳的是參數
    # session.query(Dep.id).filter(Dep.dname=='銷售') #傳的是表達式
)).all()
形式二:子查詢當作in的範圍用,調用in_ 
#示例:查詢全部的員工姓名與部門名

#原生SQL:
# select ename as 員工姓名,(select dname from dep where id = emp.dep_id) as 部門名 from emp;

#ORM:
sub_sql=session.query(Dep.dname).filter(Dep.id==Emp.dep_id) #SELECT dep.dname FROM dep, emp WHERE dep.id = emp.dep_id
sub_sql.as_scalar() #as_scalar的功能就是把上面的sub_sql加上了括號

res=session.query(Emp.ename,sub_sql.as_scalar()).all()
形式三:子查詢當作select後的字段,調用as_scalar()

 

#!/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 sqlalchemy.sql import text, func
from sqlalchemy.engine.result import ResultProxy
from db import Users, Hosts, Hobby, Person, Group, Server, Server2Group

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/demo?charset=utf8", 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`
"""


# 原生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)
"""

session.close()
關聯子查詢
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程