flask的SQLAlchemy，连接数据库的增删改查操作

flask的SQLAlchemy

"""
	安装：
    pip install  sqlalchemy
"""

models.py

# pip install  sqlalchemy
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 Users(Base):
    __tablename__ = 'users'  # 数据库表名称
    id = Column(Integer, primary_key=True)  # id 主键
    name = Column(String(32), index=True, nullable=False)  # name列，索引，不可为空
    age = Column(Integer)
    #email = Column(String(32), unique=True)
    #datetime.datetime.now#不能加括号，加了括号，以后永远是当前时间
    #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'), #索引
    )

    def __repr__(self):  # 作用类似于__str__, 但是__repr__的作用更加底层，__str__则打印不出来
        return self.name


def init_db():
    """
    根据类创建数据库表
    :return:
    """
    engine = create_engine(
        "mysql+pymysql://root:123456@127.0.0.1:3306/python13?charset=utf8",
        #"什么数据库（mysql,orcal）+用什么取链接数据库（pymysql）://数据库用户名:密码@mysqlip:端口/数据库名？charset=字符集"
        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:123456@127.0.0.1:3306/python13?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()

orm的操作去连接数据库

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Users

# 创造你的数据库引擎
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/python13", max_overflow=0, pool_size=5)

# 绑定引擎
Connection = sessionmaker(bind=engine) 

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

con = Connection()

obj1 = Users(name='lsb1', age=28)
con.add(obj1)


# 必须提交之后才能生效
con.commit()

# 关闭连接
con.close()

orm的增删改查，F查询，以及SQL语句

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Users
#"mysql+pymysql://root@127.0.0.1:3306/aaa"
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/python13", max_overflow=0, pool_size=5)
Connection = sessionmaker(bind=engine)

# 每次执行数据库操作时，都需要创建一个Connection
con = Connection()

# 1 单增
obj1  = Users(name="lsb1",age=12)
con.add(obj1)

# 2 多个增加

con.add_all([
        Users(name="lsb1",age=12),
        Users(name="esb",age=40),
        Users(name="jsb",age=30),
        Users(name="tsb",age=12),
        #Host(name = "tsb",time=123213)
])



# 3 删除
# 想对表做操作必须使用query
con.query(Users).delete()  # query（表）后面不加任何东西就是SQL语句


#4 改
con.query(Users).update({"name":"sb","age":14})

# 类似django的F查询, 如果你修改的是一个字符串类型，得在后面加上synchronize_session=False
con.query(Users).update({Users.name:Users.name +" is true","age":1},synchronize_session=False)


# 如果是数字类型则不需要设置synchronize_session=False，当然你设置也没关系
con.query(Users).update({Users.age:Users.age + 10})


# 5查(查是不需要commit,也能拿到结果)
#打印sql
r1 = con.query(Users)

#查询所有
r1 = con.query(Users).all()
#
#查单条记录
r1 = con.query(Users).first()

#查哪些字段
r1 = con.query(Users.age,Users.name.label("sb")).first()
# .label是取别名


#过滤用filter_by(传参数)或者filter（传表达式）
r1 = con.query(Users).filter(Users.name == "sb").first()
# 可多条一起操作
con.query(Users).filter(Users.name == "tsb").update({"name": "sb", "age": 14})


r1 =  con.query(Users).filter_by(name = "esb").first()
#
# print(r1.sb) # 如果你取了别名以后你想查名字，只能点你的别名查找

print(r1)
#必须提交才能生效
con.commit()

#关闭链接
con.close()