ICode9

精准搜索请尝试: 精确搜索
首页 > 数据库> 文章详细

0221 SQLAlchemy

2020-03-10 22:01:21  阅读:296  来源: 互联网

标签:Tag Users 0221 session query SQLAlchemy id name


目录

SQlAlchemy

介绍

SQLAlchemy是Python编程语言下的一款ORM框架,该框架建立在数据库API之上,使用关系对象映射进行数据库操作,简言之便是:将对象转换成SQL,然后使用数据API执行SQL并获取执行结果。

补充

MySQL数据库连接不上的原因

# 1 用户名密码错误,用户权限不够,
# 2 ip和端口不对。远程ip,该服务器的防火墙有没有关闭,端口不一定是3306,3307,3308,3309,3310

ORM

创建表

# 1. 创建单表
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy import Index, UniqueConstraint
import datetime

ENGINE = create_engine("mysql+pymysql://root:123abc@127.0.0.1:3306/test?charset=utf8mb4",)

# Base是declarative_base的实例化对象
Base = declarative_base()


# 每个类都要继承Base
class UserInfo(Base):
    # __tablename__是必须要的,它是设置实际存在数据库中的表名
    __tablename__ = "user_info"

    # Column是列的意思,固定写法 Column(字段类型, 参数)
    # primary_key主键、index索引、nullable是否可以为空
    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False)
    email = Column(String(32), unique=True)
    create_time = Column(DateTime, default=datetime.datetime.now)

    # 相当于Django的ORM的class Meta,是一些元信息
    __table_args__ = (
        UniqueConstraint("id", "name", name="uni_id_name"),
        Index("name", "email")
    )


def create_db():
    # metadata.create_all创建所有表
    Base.metadata.create_all(ENGINE)


def drop_db():
    # metadata.drop_all删除所有表
    Base.metadata.drop_all(ENGINE)


if __name__ == '__main__':
    create_db()

一对一表

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy import Index, UniqueConstraint, ForeignKey
from sqlalchemy.orm import relationship
import datetime


ENGINE = create_engine("mysql+pymysql://root:123abc@127.0.0.1:3306/test?charset=utf8mb4",)

Base = declarative_base()


# ======一对多示例=======
class UserInfo(Base):
    __tablename__ = "user_info"

    id = Column(Integer, primary_key=True)
    # index=True,设置索引
    name = Column(String(32), index=True, nullable=False)
    email = Column(String(32), unique=True)
    create_time = Column(DateTime, default=datetime.datetime.now)
    # ForeignKey字段的建立,需要指定外键绑定哪个表的哪个字段
    hobby_id = Column(Integer, ForeignKey("hobby.id"))
    # 不生成表结构 方便查询和增加的操作
    # 第一个参数是关联到哪个类(表), backref是给关联的那个类反向查询用的
    hobby = relationship("Hobby", backref="user")

    __table_args__ = (
        # UniqueConstraint联合唯一,这个联合唯一的字段名为:uni_id_name
        UniqueConstraint("id", "name", name="uni_id_name"),
        # 联合索引
        Index("name", "email")
    )


class Hobby(Base):
    __tablename__ = "hobby"

    id = Column(Integer, primary_key=True)
    title = Column(String(32), default="码代码")




def create_db():
    Base.metadata.create_all(ENGINE)


def drop_db():
    Base.metadata.drop_all(ENGINE)



if __name__ == '__main__':
    create_db()
    # drop_db()

一对多关系

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指的是tablename而不是类名,uselist=False
    hobby_id = Column(Integer, ForeignKey("hobby.id"))
    
    # 跟数据库无关,不会新增字段,只用于快速链表操作
    # 类名,backref用于反向查询
    hobby=relationship('Hobby',backref='pers')

多对多表

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy import Index, UniqueConstraint, ForeignKey
from sqlalchemy.orm import relationship
import datetime


ENGINE = create_engine("mysql+pymysql://root:123abc@127.0.0.1:3306/test?charset=utf8mb4",)

Base = declarative_base()


# ======多对多示例=======
class Book(Base):
    __tablename__ = "book"

    id = Column(Integer, primary_key=True)
    title = Column(String(32))
    # 不生成表字段 仅用于查询和增加方便
    # 多对多的relationship还需要设置额外的参数secondary:绑定多对多的中间表
    tags = relationship("Tag", secondary="book2tag", backref="books")


class Tag(Base):
    __tablename__ = "tag"

    id = Column(Integer, primary_key=True)
    title = Column(String(32))


class Book2Tag(Base):
    __tablename__ = "book2tag"

    id = Column(Integer, primary_key=True)
    book_id = Column(Integer, ForeignKey("book.id"))
    tag_id = Column(Integer, ForeignKey("tag.id"))


def create_db():
    Base.metadata.create_all(ENGINE)

def drop_db():
    Base.metadata.drop_all(ENGINE)

if __name__ == '__main__':
    create_db()
    # drop_db()

数据库操作

scoped_session

# 1. scoped_session
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from models_demo import Tag


ENGINE = create_engine("mysql+pymysql://root:123abc@127.0.0.1:3306/test?charset=utf8mb4",)

# 每次执行数据库操作的时候,都需要创建一个session,相当于管理器(相当于Django的ORM的objects)
Session = sessionmaker(bind=ENGINE)
# 线程安全,基于本地线程实现每个线程用同一个session
session = scoped_session(Session)


# =======执行ORM操作==========
tag_obj = Tag(title="SQLAlchemy")
# 添加
session.add(tag_obj)
# 提交
session.commit()
# 关闭session
session.close()

增删改查

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from models_demo import Tag, UserInfo
import threading


ENGINE = create_engine("mysql+pymysql://root:123abc@127.0.0.1:3306/test?charset=utf8mb4",)

Session = sessionmaker(bind=ENGINE)

# 每次执行数据库操作的时候,都需要创建一个session
session = Session()
session = scoped_session(Session)


# ============添加================
tag_obj = Tag(title="SQLAlchemy")
session.add(tag_obj)

# 批量添加
session.add_all([
    Tag(title="Python"),
    Tag(title="Django"),
])
# 提交
session.commit()
# 关闭session
session.close()


# ============基础查询============
ret = session.query(Tag).all()
# get(id)
ret1 = session.query(Tag).get(1)  # 查询Tag表 id=1的记录
# filter(表达式)
ret2 = session.query(Tag).filter(Tag.title == "Python").all()
# filter_by(字段=xx)
ret3 = session.query(Tag).filter_by(title="Python").all()
ret4 = session.query(Tag).filter_by(title="Python").first()
print(ret1, ret2, ret3, ret4)


# ============删除===========
session.query(Tag).filter_by(id=1).delete()
session.commit()


# ===========修改===========
session.query(Tag).filter_by(id=22).update({Tag.title: "LOL"})
session.query(Tag).filter_by(id=23).update({"title": "吃鸡"})
session.query(Tag).filter_by(id=24).update({"title": Tag.title + "~"}, synchronize_session=False)
# synchronize_session="evaluate" 默认值进行数字加减
session.commit()

常用操作

# 条件查询
ret1 = session.query(Tag).filter_by(id=22).first()
ret2 = session.query(Tag).filter(Tag.id > 1, Tag.title == "LOL").all()
ret3 = session.query(Tag).filter(Tag.id.between(22, 24)).all()
ret4 = session.query(Tag).filter(~Tag.id.in_([22, 24])).first()
from sqlalchemy import and_, or_
ret5 = session.query(Tag).filter(and_(Tag.id > 1, Tag.title == "LOL")).first()
ret6 = session.query(Tag).filter(or_(Tag.id > 1, Tag.title == "LOL")).first()
ret7 = session.query(Tag).filter(or_(
    Tag.id>1,
    and_(Tag.id>3, Tag.title=="LOL")
)).all()

# 通配符
ret8 = session.query(Tag).filter(Tag.title.like("L%")).all()
ret9 = session.query(Tag).filter(~Tag.title.like("L%")).all()

# 限制
ret10 = session.query(Tag).filter(~Tag.title.like("L%")).all()[1:2]

# 排序
ret11 = session.query(Tag).order_by(Tag.id.desc()).all()  # 倒序
ret12 = session.query(Tag).order_by(Tag.id.asc()).all()  # 正序

# 分组
ret13 = session.query(Tag.test).group_by(Tag.test).all()

# 聚合函数
from sqlalchemy.sql import func
ret14 = session.query(
    func.max(Tag.id),
    func.sum(Tag.test),
    func.min(Tag.id)
).group_by(Tag.title).having(func.max(Tag.id > 22)).all()

# 连表
# print(ret15) 得到一个列表套元组 元组里是两个对象
# [(user_obj1, hobby_obj1), (user_obj2, hobby_obj2), ]
ret15 = session.query(UserInfo, Hobby).filter(UserInfo.hobby_id == Hobby.id).all()

# print(ret16) 得到列表里面是前一个对象,join相当于inner join
# [user_obj1, user_obj2, ]
ret16 = session.query(UserInfo).join(Hobby).all()

# 相当于inner join
# for i in ret16:
#     # print(i[0].name, i[1].title)
#     print(i.hobby.title)

# 指定isouter=True相当于left join
ret17 = session.query(Hobby).join(UserInfo, isouter=True).all()
ret17_1 = session.query(UserInfo).join(Hobby, isouter=True).all()

# 或者直接用outerjoin也是相当于left join
ret18 = session.query(Hobby).outerjoin(UserInfo).all()
ret18_1 = session.query(UserInfo).outerjoin(Hobby).all()
print(ret17)
print(ret17_1)
print(ret18)
print(ret18_1)




------------------------------------------------------------



# 条件
ret = session.query(Users).filter_by(name='lqz').all()
#表达式,and条件连接
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()
#注意下划线
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_
#or_包裹的都是or条件,and_包裹的都是and条件
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()


# 通配符,以e开头,不以e开头
ret = session.query(Users).filter(Users.name.like('e%')).all()
ret = session.query(Users).filter(~Users.name.like('e%')).all()

# 限制,用于分页,区间
ret = session.query(Users)[1:2]

# 排序,根据name降序排列(从大到小)
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()
#分组之后取最大id,id之和,最小id
ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).all()
#haviing筛选
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()

# 连表(默认用forinkey关联)

ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()
#join表,默认是inner join
ret = session.query(Person).join(Favor).all()
#isouter=True 外连,表示Person left join Favor,没有右连接,反过来即可
ret = session.query(Person).join(Favor, isouter=True).all()
#打印原生sql
aa=session.query(Person).join(Favor, isouter=True)
print(aa)
# 自己指定on条件(连表条件),第二个参数,支持on多个条件,用and_,同上
ret = session.query(Person).join(Favor,Person.id==Favor.id, isouter=True).all()
# 组合(了解)UNION 操作符用于合并两个或多个 SELECT 语句的结果集
#union和union 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的ForeignKey

# 4. 基于relationship的ForeignKey
# 添加
user_obj = UserInfo(name="提莫", hobby=Hobby(title="种蘑菇"))
session.add(user_obj)

hobby = Hobby(title="弹奏一曲")
hobby.user = [UserInfo(name="琴女"), UserInfo(name="妹纸")]
# hobby.user = [session.query(UserInfo).filter_by(id=1).first(), ]
session.add(hobby)
session.commit()

# 基于relationship的正向查询
user_obj_1 = session.query(UserInfo).first()
print(user_obj_1.name)
print(user_obj_1.hobby.title)

# 基于relationship的反向查询
hb = session.query(Hobby).first()
print(hb.title)
for i in hb.user:
    print(i.name)

session.close()

基于relationship的M2M

# 5. 基于relationship的M2M
# 添加
# 直接给中间表添加
book_obj = Book(title="Python源码剖析")
tag_obj = Tag(title="Python")
b2t = Book2Tag(book_id=book_obj.id, tag_id=tag_obj.id)
session.add_all([
    book_obj,
    tag_obj,
    b2t,
])
session.commit()

# 通过反向字段添加
book = Book(title="测试")
book.tags = [Tag(title="测试标签1"), Tag(title="测试标签2")]
# book.tags = [session.query(Tag).filter_by(id=1).first(), ]
session.add(book)
session.commit()

tag = Tag(title="LOL")
tag.books = [Book(title="大龙刷新时间"), Book(title="小龙刷新时间")]
session.add(tag)
session.commit()

# 基于relationship的正向查询
book_obj = session.query(Book).filter_by(id=4).first()
print(book_obj.title)
print(book_obj.tags)
# 基于relationship的反向查询
tag_obj = session.query(Tag).first()
print(tag_obj.title)
print(tag_obj.books)

基于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=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'
)
"""
#scoped_session类并没有继承Session,但是却又它的所有方法
session = scoped_session(Session)
# ############# 执行ORM操作 #############
obj1 = Users(name="alex1")
session.add(obj1)

# 提交事务
session.commit()
# 关闭session
session.close()

其他

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=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()

# 关联子查询:correlate(Group)表示跟Group表做关联,as_scalar相当于对该sql加括号,用于放在后面当子查询
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`
"""
'''

select * from tb where id in [select id from xxx];

select id,
        name,
        #必须保证此次查询只有一个值
        (select max(id) from xxx) as mid
from tb

例如,第三个字段只能有一个值
id name  mid
1  lqz   1,2  不合理
2  egon   2


'''
'''
成绩表:
id sid    cid    score
1  1      物理      99 
2  1      化学      88
3  2      物理      95

学生表:
id   name  每个学生总分数
1     xx      88
2     yy       77

select id,name,
(select avr(score) from 成绩表 where 成绩表.sid=学生表.id) as x
from 学生表
subqry = session.query(func.count(成绩表.scort).label("sc")).filter(学生表.id == 成绩表.sid).correlate(学生表).as_scalar()
result = session.query(学生表.name, subqry)

'''

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

Flask-SQLAlchemy

flask和SQLAlchemy的管理者,通过他把他们做连接

db = SQLAlchemy()
    - 包含配置
    - 包含ORM基类
    - 包含create_all
    - engine
    - 创建连接

离线脚本,创建表

flask-migrate
python3 manage.py db init 初始化: 只执行一次
python3 manage.py db migrate 等同于makemigrations
python3 manage.py db upgrade 等同于 migrate

标签:Tag,Users,0221,session,query,SQLAlchemy,id,name
来源: https://www.cnblogs.com/fwzzz/p/12459067.html

本站声明: 1. iCode9 技术分享网(下文简称本站)提供的所有内容,仅供技术学习、探讨和分享;
2. 关于本站的所有留言、评论、转载及引用,纯属内容发起人的个人观点,与本站观点和立场无关;
3. 关于本站的所有言论和文字,纯属内容发起人的个人观点,与本站观点和立场无关;
4. 本站文章均是网友提供,不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属;如您发现该文章侵犯了您的权益,可联系我们第一时间进行删除;
5. 本站为非盈利性的个人网站,所有内容不会用来进行牟利,也不会利用任何形式的广告来间接获益,纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

专注分享技术,共同学习,共同进步。侵权联系[81616952@qq.com]

Copyright (C)ICode9.com, All Rights Reserved.

ICode9版权所有