python 操作数据库 --- sqlalchemy增删改查

sqlalchemy 操作数据库代码示例，具体见代码及注释。
sqlalchemy 官方文档：https://www.osgeo.cn/sqlalchemy/core/tutorial.html

#!/usr/bin/python3
# sqlalchemy 连接数据库
# pip install sqlalchemy

from sqlalchemy.orm import sessionmaker
from sqlalchemy import Integer, Column, String, ForeignKey, DateTime, create_engine, update
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

Base = declarative_base()

class Book_Table(Base):
    __tablename__ = 'bookorm'
    book_id = Column(Integer(), primary_key=True)
    book_name = Column(String(50), index=True)

    def __repr__(self):
        return 'Book_table(book_id="{self.book_id}", ' \
               'book_name={self.book_name})'.format(self=self)

class Author_table(Base):
    __tablename__ = 'authororm'
    user_id = Column(Integer(), primary_key=True)
    username = Column(String(15), nullable=False, unique=True)
    created_on = Column(DateTime(), default=datetime.now)
    updated_on = Column(DateTime(), default=datetime.now, onupdate=datetime.now)

dburl = 'mysql+pymysql://testuser:testpass@192.168.246.129:3306/testdb?charset=utf8mb4'
engine = create_engine(dburl, echo=True, encoding='utf-8')
# 创建表格
# Base.metadata.create_all(engine)

#创建会话， 创建事务，直到 session.commit()
SessionClass = sessionmaker(bind=engine)
session = SessionClass()

#增加数据
# book_demo = Book_Table(book_name='肖申克的救赎')
# book_demo1 = Book_Table(book_name='活着')
# book_demo2 = Book_Table(book_name='红楼梦')
#
# session.add(book_demo)
# session.add(book_demo1)
# session.add(book_demo2)

#查询数据， 默认 SELECT *
# result = session.query(Book_Table).all()
# result = session.query(Book_Table).first()
# .one()  结果必须只有一个，否则会报错
# 。scalar() 结果为一个或None， 否则会报错。

# 全部查询用如下方式，而不是 .all()
# for result in session.query(Book_Table):
#     print(result)

# 指定列数
# result = session.query(Book_Table.book_name).first()

# 排序
from sqlalchemy import desc
# for result in session.query(Book_Table.book_name, Book_Table.book_id).order_by(desc(Book_Table.book_id)):
#     print(result)

# 限制返回行数
# query = session.query(Book_Table).order_by(desc(Book_Table.book_id)).limit(3)
# print([result.book_name for result in query])

# 各种函数调用，例如 聚合函数，具体可以有哪些函数参照：
# https://www.osgeo.cn/sqlalchemy/core/functions.html
from sqlalchemy import func
result = session.query(func.count(Book_Table.book_name)).first()
print(result)

# 条件查询， WHERE 语句，这里是 filter。
print(session.query(Book_Table).filter(Book_Table.book_id < 20).first())
# filter(Book_table.book_id > 10, Book_table.book_id < 20)
# 与， 或 非
from sqlalchemy import and_, or_, not_
# filter(
#     or_(
#         Book_Table.xxx.between(100, 1000),
#         Book_Table.yyy.contains('book')
#     )
# )

# 更新  update()
query = session.query(Book_Table)
query = query.filter(Book_Table.book_id == 3)
query.update({Book_Table.book_name: 'newbook'})
new_book = query.first()
print(new_book)

# 删除 delete()
# ！！！ 警告，操作不可撤回，注意删除内容。
query = session.query(Book_Table)
query = query.filter(Book_Table.book_id == 1)
# result = session.delete(query.one())      # 一种删除方法，  返回 None
query.delete()     # 另一种删除方法
print(result)

session.commit()