express+node搭建自己的本地後臺
阿新 • • 發佈：2021-07-26
目錄
1. 常用的sqlalchemy欄位型別
| 型別名 | python中型別 | 說明 |
|---|---|---|
| Integer | int | 普通整數，一般是32位 |
| SmallInteger | int | 取值範圍小的整數，一般是16位 |
| BigInteger | int或long | 不限制精度的整數 |
| Float | float | 浮點數 |
| Numeric | decimal.Decimal |
普通整數，一般是32位 |
| String | str | 變長字串 |
| Text | str | 變長字串，對較長或不限長度的字串做了優化 |
| Unicode | unicode | 變長Unicode字串 |
| UnicodeText | unicode | 變長Unicode字串，對較長或不限長度的字串做了優化 |
| Boolean | bool | 布林值 |
| Date | datetime.date | 時間 |
| Time | datetime.datetime | 日期和時間 |
| LargeBinary | str | 二進位制檔案 |
2. 常用的Column選項
| 選項名 | 說明 |
|---|---|
| primary_key | 如果為True，代表表的主鍵 |
| unique | 如果為True，代表這列不允許出現重複的值 |
| index | 如果為True，為這列建立索引，提高查詢效率 |
| nullable | 如果為True，允許有空值，如果為False，不允許有空值 |
| default | 為這列定義預設值 |
| comment | 欄位註釋 |
3. 常用的關係選項
| 選項名 | 說明 |
|---|---|
| backref | 在關係的另一模型中新增反向引用 |
| primaryjoin | 明確指定兩個模型之間使用的聯結條件 |
| uselist | 如果為False，不使用列表，而使用標量值 |
| order_by | 指定關係中記錄的排序方式 |
| secondary | 指定多對多中記錄的排序方式 |
4. 簡單操作
安裝sqlalchemy模組
pip install sqlalchemy
4.1 連線資料庫並建立表
# 匯入模組
from sqlalchemy import create_engine, Column, Integer, String, DATETIME
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 建立物件的基類
Base = declarative_base()
# 定義表物件
class Users(Base):
# 表的名字
__tablename__ = "user"
# 表的結構
id = Column(Integer, primary_key=True)
username = Column(String(64), unique=True, nullable=False, comment="使用者名稱")
password = Column(String(64), unique=True, nullable=False, comment="密碼")
def __init__(self, id, username, password, create_time, update_time):
self.id = id
self.username = username
self.password = password
# 初始化資料庫連線
engine = create_engine("mysql+pymysql://root:[email protected]/system_user?charset=utf8mb4", echo=True)
# 建立DBsession型別
DBSession = sessionmaker(bind=engine)
# 建立定義的表
Base.metadata.create_all(engine)
create_engine()用來初始化資料庫連線。SQLAlchemy用一個字串表示連線資訊：
'資料庫型別+資料庫驅動名稱://使用者名稱:口令@機器地址:埠號/資料庫名'
Engine內部維護了一個Pool（連線池）和Dialect（方言），方言來識別具體連線資料庫種類。
create_engine還有其它可選的引數，比如：
engine = create_engine("mysql+pymysql://root:[email protected]/system_user?charset=utf8mb4",
echo=True,
pool_size=5,
pool_recycle=-1,
query_cache_size=500)
# echo=True:當設定為True時會將orm語句轉化為sql語句列印，一般debug的時候可用
# pool_size:連線池的大小，預設為5個，設定為0時表示連線無限制
# pool_recycle:設定時間以限制資料庫多久沒連線自動斷開,預設-1沒有限制
# query_cache_size：用於快取 SQL 字串的快取大小查詢的形式。設定為零以禁用快取。
4.2 插入資料
engine = create_engine("mysql+pymysql://root:[email protected]/system_user?charset=utf8mb4", echo=True)
DBSession = sessionmaker(bind=engine)
session = DBSession()
# 定義新增的資料
user_add = Users(
id=1,
username="abc",
password="123456"
)
# 利用session.add()向資料庫新增
session.add(user_add)
session.commit()
session.close()
4.3 查詢資料
user_message = session.query(Users).filter_by(id=1).all()
# user_message = session.query(Users).filter_by(Users.id == 1).all()
for i in user_message:
print(i.username)
# 通過session.query()我們查詢返回了一個Query物件，此時還沒有去具體的資料庫中查詢，只有當執行具體的.all()，.first()等函式時才會真的去操作資料庫。
| filter | filter_by |
|---|---|
| 支援所有比較運算子，相等比較用比較用== | 只能使用"="，"!="和"><" |
| 過濾用類名.屬性名 | 過濾用屬性名 |
| 不支援組合查詢，只能連續呼叫filter變相實現 | 引數是**kwargs，支援組合查詢 |
| 支援and，or和in等 |
4.4 更改資料
session.query(Users).filter_by(id=1).update({"username": "ccc"})
# 或者
user = session.query(Users).filter_by(id=1).first()
user.username = "aaa"
session.add(user)
4.5 刪除資料
delete_user = session.query(Users).filter_by(id=1).first()
if delete_user:
session.delete(delete_user)
session.commit()
# 或者
session.query(Users).filter_by(id=1).delete()
session.commit()
5. 特殊操作
5.1 時間列自動更新
from sqlalchemy import create_engine, Column, Integer, String, TIMESTAMP, text
...
create_time = Column(TIMESTAMP, server_default=text("CURRENT_TIMESTAMP"), comment="建立時間", nullable=False)
update_time = Column(TIMESTAMP, server_default=text("CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"), comment="更新時間", nullable=False)
...
5.2 批量新增
# 第一種方法
user_add_list = [
Users(id=1, username="aaa", password="1", create_time=None, update_time=None),
Users(id=2, username="bbb", password="2", create_time=None, update_time=None),
Users(id=3, username="ccc", password="3", create_time=None, update_time=None),
Users(id=4, username="ddd", password="4", create_time=None, update_time=None),
]
session.add_all(user_add_list)
session.commit()
# 第二種方法
user_add_list = [
Users(id=1, username="aaa", password="1", create_time=None, update_time=None),
Users(id=2, username="bbb", password="2", create_time=None, update_time=None),
Users(id=3, username="ccc", password="3", create_time=None, update_time=None),
Users(id=4, username="ddd", password="4", create_time=None, update_time=None),
]
session.bulk_save_objects(user_add_list)
session.commit()
5.3 查詢引數
# 返回所有符合條件的記錄
session.query(Users).filter_by(id=1).all()
# 返回符合條件的第一條記錄
session.query(Users).filter_by(id=1).first()
# 返回一個物件，如果有多個報錯
session.query(Users).filter_by(Users.id=1).one()
# 限制返回條數
session.query(Users).filter(Users.id=1).limit(2).all()
# 從第3條開始返回記錄
session.query(Users).filter(Users.id=1).offset(2).all()
# 切片返回記錄
session.query(Users).filter(Users.id=1).slice(2,3).all()
# 按順序排序返回記錄
session.query(Users).filter(Users.username="abc").order_by(id).all()
# 降序返回記錄
session.query(Users).filter(Users.username="abc").order_by(desc(id)).all()
# 不等於
session.query(Users).filter(Users.id!=1).all()
# 模糊匹配
session.query(Users).filter(Users.username.like("aaa")).all()
session.query(Users).filter(Users.username.notlike("aaa")).all()
# 成員屬於/不屬於
session.query(User).filter(User.username.in_(['aaa','bbb'])).all()
session.query(User).filter(User.username.notin_(['aaa','bbb'])).all()
# 空判斷
session.query(User).filter(User.username==None).all()
session.query(User).filter(User.username.is_(None)).all()
session.query(User).filter(User.username.isnot(None)).all()
# 多條件
session.query(User).filter(User.username.isnot(None),User.password=='123').all()
# and or
session.query(User).filter(or_(User.username=='aaa',User.password=='123')).all()
session.query(User).filter(and_(User.username=='bbb',User.password=='111')).all()
# 計算個數
session.query(Users).filter_by(id=1).count()