python 操作數(shù)據(jù)庫 --- sqlalchemy增刪改查

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()
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成,瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明:文章內(nèi)容(如有圖片或視頻亦包括在內(nèi))由作者上傳并發(fā)布,文章內(nèi)容僅代表作者本人觀點,簡書系信息發(fā)布平臺,僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容