SQLAlchemy(3)

阅读原文时间：2023年07月11日阅读：1

一. 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://:@[:]/

pymysql
mysql+pymysql://:@/[?]

MySQL-Connector
mysql+mysqlconnector://:@[:]/

cx_Oracle
oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value…]

1. 执行原生SQL语句

举例1

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=, # 超过连接池大小外最多创建的连接

pool_size=, # 连接池大小

pool_timeout=, # 池中没有线程最多等待的时间，否则报错

pool_recycle=- # 多久之后对线程池中的线程进行一次连接的回收（重置）

)

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

t = threading.Thread(target=task, args=(i,))

t.start()

举例2

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=, pool_size=)

def task(arg):

conn = engine.contextual_connect()

with conn:

cur = conn.execute(

"select * from t1"

)

result = cur.fetchall()

print(result)

for i in range():

t = threading.Thread(target=task, args=(i,))

t.start()

举例3

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=, pool_size=)

def task(arg):

cur = engine.execute("select * from t1")

result = cur.fetchall()

cur.close()

print(result)

for i in range():

t = threading.Thread(target=task, args=(i,))

t.start()

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

2. ORM

a. 创建数据库表

创建单表

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 </span>= Column(Integer, primary_key=<span style="color: #000000;">True)
name </span>= Column(String(<span style="color: #800080;">32</span>), index=True, nullable=<span style="color: #000000;">False)
# email </span>= Column(String(<span style="color: #800080;">32</span>), unique=<span style="color: #000000;">True)
# ctime </span>= Column(DateTime, <span style="color: #0000ff;">default</span>=<span style="color: #000000;">datetime.datetime.now)
# extra </span>= Column(Text, nullable=<span style="color: #000000;">True)

__table_args__ </span>=<span style="color: #000000;"> (
    # UniqueConstraint(</span><span style="color: #800000;">'</span><span style="color: #800000;">id</span><span style="color: #800000;">'</span>, <span style="color: #800000;">'</span><span style="color: #800000;">name</span><span style="color: #800000;">'</span>, name=<span style="color: #800000;">'</span><span style="color: #800000;">uix_id_name</span><span style="color: #800000;">'</span><span style="color: #000000;">),
    # Index(</span><span style="color: #800000;">'</span><span style="color: #800000;">ix_id_name</span><span style="color: #800000;">'</span>, <span style="color: #800000;">'</span><span style="color: #800000;">name</span><span style="color: #800000;">'</span>, <span style="color: #800000;">'</span><span style="color: #800000;">email</span><span style="color: #800000;">'</span><span style="color: #000000;">),
)

def init_db():

"""

根据类创建数据库表

:return:

"""

engine = create\_engine(

"mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",

max_overflow=, # 超过连接池大小外最多创建的连接

pool_size=, # 连接池大小

pool_timeout=, # 池中没有线程最多等待的时间，否则报错

pool_recycle=- # 多久之后对线程池中的线程进行一次连接的回收（重置）

)

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=, # 超过连接池大小外最多创建的连接

pool_size=, # 连接池大小

pool_timeout=, # 池中没有线程最多等待的时间，否则报错

pool_recycle=- # 多久之后对线程池中的线程进行一次连接的回收（重置）

)

Base.metadata.drop_all(engine)

if name == 'main':

drop_db()

init_db()

创建单表

创建多个表并包含Fk、M2M关系

Base = declarative_base()

##################### 单表示例

class Users(Base):

tablename = 'users'

id </span>= Column(Integer, primary_key=<span style="color: #000000;">True)
name </span>= Column(String(<span style="color: #800080;">32</span>), index=<span style="color: #000000;">True)
age </span>= Column(Integer, <span style="color: #0000ff;">default</span>=<span style="color: #800080;">18</span><span style="color: #000000;">)
email </span>= Column(String(<span style="color: #800080;">32</span>), unique=<span style="color: #000000;">True)
ctime </span>= Column(DateTime, <span style="color: #0000ff;">default</span>=<span style="color: #000000;">datetime.datetime.now)
extra </span>= Column(Text, nullable=<span style="color: #000000;">True)

__table_args__ </span>=<span style="color: #000000;"> (
    # UniqueConstraint(</span><span style="color: #800000;">'</span><span style="color: #800000;">id</span><span style="color: #800000;">'</span>, <span style="color: #800000;">'</span><span style="color: #800000;">name</span><span style="color: #800000;">'</span>, name=<span style="color: #800000;">'</span><span style="color: #800000;">uix_id_name</span><span style="color: #800000;">'</span><span style="color: #000000;">),
    # Index(</span><span style="color: #800000;">'</span><span style="color: #800000;">ix_id_name</span><span style="color: #800000;">'</span>, <span style="color: #800000;">'</span><span style="color: #800000;">name</span><span style="color: #800000;">'</span>, <span style="color: #800000;">'</span><span style="color: #800000;">extra</span><span style="color: #800000;">'</span><span style="color: #000000;">),
)

class Hosts(Base):

tablename = 'hosts'

id </span>= Column(Integer, primary_key=<span style="color: #000000;">True)
name </span>= Column(String(<span style="color: #800080;">32</span>), index=<span style="color: #000000;">True)
ctime </span>= Column(DateTime, <span style="color: #0000ff;">default</span>=<span style="color: #000000;">datetime.datetime.now)

##################### 一对多示例

class Hobby(Base):

tablename = 'hobby'

id = Column(Integer, primary_key=True)

caption = Column(String(), default='篮球')

class Person(Base):

tablename = 'person'

nid = Column(Integer, primary_key=True)

name = Column(String(), index=True, nullable=True)

hobby_id = Column(Integer, ForeignKey("hobby.id"))

# 与生成表结构无关，仅用于查询方便
hobby </span>= relationship(<span style="color: #800000;">"</span><span style="color: #800000;">Hobby</span><span style="color: #800000;">"</span>, backref=<span style="color: #800000;">'</span><span style="color: #800000;">pers</span><span style="color: #800000;">'</span><span style="color: #000000;">)

##################### 多对多示例

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(), unique=True, nullable=False)

# 与生成表结构无关，仅用于查询方便
servers </span>= relationship(<span style="color: #800000;">'</span><span style="color: #800000;">Server</span><span style="color: #800000;">'</span>, secondary=<span style="color: #800000;">'</span><span style="color: #800000;">server2group</span><span style="color: #800000;">'</span>, backref=<span style="color: #800000;">'</span><span style="color: #800000;">groups</span><span style="color: #800000;">'</span><span style="color: #000000;">)

class Server(Base):

tablename = 'server'

id </span>= Column(Integer, primary_key=True, autoincrement=<span style="color: #000000;">True)
hostname </span>= Column(String(<span style="color: #800080;">64</span>), unique=True, nullable=<span style="color: #000000;">False)

def init_db():

"""

根据类创建数据库表

:return:

"""

engine = create\_engine(

"mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",

max_overflow=, # 超过连接池大小外最多创建的连接

pool_size=, # 连接池大小

pool_timeout=, # 池中没有线程最多等待的时间，否则报错

pool_recycle=- # 多久之后对线程池中的线程进行一次连接的回收（重置）

)

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=, # 超过连接池大小外最多创建的连接

pool_size=, # 连接池大小

pool_timeout=, # 池中没有线程最多等待的时间，否则报错

pool_recycle=- # 多久之后对线程池中的线程进行一次连接的回收（重置）

)

Base.metadata.drop_all(engine)

if name == 'main':

drop_db()

init_db()

创建多个表并包含Fk、M2M关系

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

b. 操作数据库表

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=, pool_size=)

Session = sessionmaker(bind=engine)

每次执行数据库操作时，都需要创建一个session

session = Session()

############# 执行ORM操作

obj1 = Users(name="alex1")

session.add(obj1)

提交事务

session.commit()

关闭session

session.close()

基于scoped_session实现线程安全

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=, pool_size=)

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实现线程安全

多线程执行示例

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=, pool_size=)

Session = sessionmaker(bind=engine)

def task(arg):

session = Session()

obj1 </span>= Users(name=<span style="color: #800000;">"</span><span style="color: #800000;">alex1</span><span style="color: #800000;">"</span><span style="color: #000000;">)
session.add(obj1)

session.commit()

for i in range():

t = threading.Thread(target=task, args=(i,))

t.start()

基本增删改查示例

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/s6", max_overflow=, pool_size=)

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 > ).delete()

session.commit()

"""

################ 修改 ################

"""

session.query(Users).filter(Users.id > ).update({"name" : ""})

session.query(Users).filter(Users.id > ).update({Users.name: Users.name + ""}, synchronize_session=False)

session.query(Users).filter(Users.id > ).update({"age": Users.age + }, 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=, 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()

基本增删改查示例

常用操作

#　条件
ret = session.query(Users).filter_by(name='alex').all()
ret = session.query(Users).filter(Users.id > , Users.name == 'eric').all()
ret = session.query(Users).filter(Users.id.between(, ), Users.name == 'eric').all()
ret = session.query(Users).filter(Users.id.in_([,,])).all()
ret = session.query(Users).filter(~Users.id.in_([,,])).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 > , Users.name == 'eric')).all()
ret = session.query(Users).filter(or_(Users.id < , Users.name == 'eric')).all() ret = session.query(Users).filter( or_( Users.id < , and_(Users.name == 'eric', Users.id > ),
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)[:]

排序

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

q2 = session.query(Favor.caption).filter(Favor.nid < )

ret = q1.union(q2).all()

q1 = session.query(Users.name).filter(Users.id > )

q2 = session.query(Favor.caption).filter(Favor.nid < )

ret = q1.union_all(q2).all()

原生SQL语句

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

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=, pool_size=)

Session = sessionmaker(bind=engine)

session = Session()

查询

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

原生SQL语句

基于relationship操作ForeignKey

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=, pool_size=)

Session = sessionmaker(bind=engine)

session = Session()

添加

"""

session.add_all([

Hobby(caption='乒乓球'),

Hobby(caption='羽毛球'),

Person(name='张三', hobby_id=),

Person(name='李四', hobby_id=),

])

person = Person(name='张九', hobby=Hobby(caption='姑娘'))

session.add(person)

hb = Hobby(caption='人妖')

hb.pers = [Person(name='文飞'), Person(name='博雅')]

session.add(hb)

session.commit()

"""

使用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操作ForeignKey

基于relationship操作m2m

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=, pool_size=)

Session = sessionmaker(bind=engine)

session = Session()

添加

"""

session.add_all([

Server(hostname='c1.com'),

Server(hostname='c2.com'),

Group(name='A组'),

Group(name='B组'),

])

session.commit()

s2g = Server2Group(server_id=, group_id=)

session.add(s2g)

session.commit()

gp = Group(name='C组')

gp.servers = [Server(hostname='c3.com'),Server(hostname='c4.com')]

session.add(gp)

session.commit()

ser = Server(hostname='c6.com')

ser.groups = [Group(name='F组'),Group(name='G组')]

session.add(ser)

session.commit()

"""

使用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()

基于relationship操作m2m

其他

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/s6?charset=utf8", max_overflow=, pool_size=)

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

https://www.cnblogs.com/wupeiqi/articles/8259356.html

手机扫一扫

移动阅读更方便

你可能感兴趣的文章

SpringMVC--从理解SpringMVC执行流程到SSM框架整合

drozer浅析三：命令实现与交互

JavaEE期末复习

uni-app学习资料整理-1.白话uni-app

fltp备份文件后统计验证

java 工具类验证码