sqlalchemy入门增删改查使用(一)

建立数据库连接

from sqlalchemy.ext.declarative import declarative_base

BaseModel = declarative_base()

# ORM
# O R M
# Class - Table

# 创建 Class / Table
from sqlalchemy import Column,Integer,String

class User(BaseModel):
    __tablename__ = "user" # 创建Table时名字
    id = Column(Integer,primary_key=True,autoincrement=True)
    name = Column(String(32),nullable=False,index=True,unique=True)
    # Column 定义数据列
    # int string 数据类型

# 数据库引擎的创建:
from sqlalchemy.engine import create_engine
engine = create_engine('mysql://user:password@localhost:3306/test?charset=utf8mb4',
   echo=False
   pool_size=100,
   pool_recycle=3600,
   pool_pre_ping=True)

'''
echo :为 True 时候会把sql语句打印出来,当然,你可以通过配置logger来控制输出,这里不做讨论。
pool_size: 是连接池的大小,默认为5个,0表示连接数无限制
pool_recycle: MySQL 默认情况下一个连接8小时,
如果内容没有任何动作(查询请求)就会自动断开链接,出现 MySQL has gone away的错误。
设置了 pool_recycle 后 SQLAlchemy 就会在指定时间内回收连接。
如果设置为3600 就表示 1小时后该连接会被自动回收。
pool_pre_ping : 这是1.2新增的参数。
如果值为True,那么每次从连接池中拿连接的时候,
都会向数据库发送一个类似 select 1 的测试查询语句来判断服务器是否正常运行。
当该连接出现 disconnect 的情况时,该连接连同pool中的其它连接都会被回收。
'''

#利用 User 去数据库创建 user Table,自动发现数据表
BaseModel.metadata.create_all(engine) # 数据库引擎

# 1.选择数据库
from sqlalchemy.engine import create_engine
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s21?charset=utf8")
# 3.创建查询窗口
from sqlalchemy.orm import sessionmaker
select_db = sessionmaker(engine) # 选中数据库
db_session = select_db() # 已经打开查询窗口
# 4.写入SQL语句
user = User(name="Alexander.DSB.Li") # == insert into user(`name`) value ("Alexander.DSB.Li")
user_list = [User(name="Alex's Father"),User(name="李杰")]
# 放入查询窗口
db_session.add(user)
db_session.add_all(user_list)
# 5.提交sql语句
db_session.commit()
# 6.关闭查询窗口
db_session.close()

删改查

# 模拟 navcat 操作
# 1.选择数据库
from sqlalchemy.engine import create_engine
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s21?charset=utf8")
# 2.选择表
# 3.创建查询窗口
from sqlalchemy.orm import sessionmaker
select_db = sessionmaker(engine) # 选中数据库
db_session = select_db() # 已经打开查询窗口

# 简单无条件查询
# """
# select * from user  table_user == class_User
# """
res = db_session.query(User).all() # 查询全部符合条件的数据
res = db_session.query(User).first() # 查询符合条件的第一条数据
print(res.id,res.name)

# 简单条件查询
# """
# select * from user where id=3
# """
res = db_session.query(User).filter(User.id==3).all()
print(res[0].id,res[0].name)
res = db_session.query(User).filter_by(id=3).all()

res = db_session.query(User).filter(User.id==3 , User.name == "123").all()
print(res)
# 注意点
# is_true_or_false = User.id==3 and User.name == "123"


# 修改数据 update
res = db_session.query(User).filter(User.id == 1).update({"name":"李亚历山大"})
# db_session.commit()
# db_session.close()

# 删除数据
res = db_session.query(User).filter(User.id == 2).delete()
db_session.commit()
db_session.close()

版权声明:除特别注明外,本站所有文章均为王晨曦个人站点原创

转载请注明:出处来自王晨曦个人站点 » sqlalchemy入门增删改查使用(一)

点赞

发表评论

电子邮件地址不会被公开。 必填项已用*标注