sqlalchemy 操作數(shù)據(jù)庫代碼示例,具體見代碼及注釋。
sqlalchemy 官方文檔:https://www.osgeo.cn/sqlalchemy/core/tutorial.html
#!/usr/bin/python3
# sqlalchemy 連接數(shù)據(jù)庫
# 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')
# 創(chuàng)建表格
# Base.metadata.create_all(engine)
#創(chuàng)建會話, 創(chuàng)建事務(wù),直到 session.commit()
SessionClass = sessionmaker(bind=engine)
session = SessionClass()
#增加數(shù)據(jù)
# 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)
#查詢數(shù)據(jù), 默認 SELECT *
# result = session.query(Book_Table).all()
# result = session.query(Book_Table).first()
# .one() 結(jié)果必須只有一個,否則會報錯
# 。scalar() 結(jié)果為一個或None, 否則會報錯。
# 全部查詢用如下方式,而不是 .all()
# for result in session.query(Book_Table):
# print(result)
# 指定列數(shù)
# 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)
# 限制返回行數(shù)
# query = session.query(Book_Table).order_by(desc(Book_Table.book_id)).limit(3)
# print([result.book_name for result in query])
# 各種函數(shù)調(diào)用,例如 聚合函數(shù),具體可以有哪些函數(shù)參照:
# 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()
# ?。?! 警告,操作不可撤回,注意刪除內(nèi)容。
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()