在開發大多數應用時,用户系統都是必不可少的部分,而我們總是需要開發圍繞用户的登錄,註冊,獲取,更新等接口。這篇文章將帶你用一百多行代碼簡潔地實現一套這樣的用户鑑權與 RESTful 接口,並使用 Session 來處理用户的登錄登出
我們將使用 UtilMeta 框架 完成接口開發,這是一個開源的 Python 後端元框架,同時支持接入與適配 Django, Flask, FastAPI 等主流 Python 框架,並且能簡潔高效地開發 RESTful 接口
0. 安裝框架
使用如下命令即可安裝 UtilMeta 框架
pip install utilmeta
UtilMeta 框架需要 Python 版本 >= 3.8
1. 創建項目
我們使用如下命令來創建一個新項目
meta setup demo-user
我們的項目將會使用 Django 作為底層框架,所以在提示選擇 backend 的時候我們可以輸入 django
項目創建好後,我們需要先對服務的數據庫連接進行配置,打開 server.py,在 service 的聲明下面插入以下代碼
service = UtilMeta(
__name__,
name='demo-user',
backend=django,
)
# new +++++
from utilmeta.core.server.backends.django import DjangoSettings
from utilmeta.core.orm import DatabaseConnections, Database
service.use(DjangoSettings(
secret_key='YOUR_SECRET_KEY',
))
service.use(DatabaseConnections({
'default': Database(
name='db',
engine='sqlite3',
)
}))
在插入的代碼中,我們聲明瞭 Django 的配置信息與數據庫連接的配置
由於 Django 使用 app (應用) 的方式來管理數據模型,接下來我們使用如下的命令來創建一個名為 user 的 app
cd demo-user
meta add user
可以看到在我們的項目文件夾中新創建出了一個 user 文件夾,其中包括
/user
/migrations
api.py
models.py
schema.py
其中 migrations 文件夾是 Django 用來處理數據庫遷移文件的,models.py 是我們編寫數據模型的地方
應用創建完成後,我們將 server.py 的 Django 設置中插入一行代碼來注入新創建的 user app
service.use(DjangoSettings(
secret_key='YOUR_SECRET_KEY',
apps=['user'] # new
))
至此我們完成了項目的配置和初始化
2. 編寫用户模型
用户的登錄註冊 API 當然是圍繞 “用户” 進行的了,在開發 API 之前,我們需要先編寫好用户的數據模型,我們打開 user/models.py,編寫
from django.db import models
from utilmeta.core.orm.backends.django.models import AbstractSession, PasswordField
class User(models.Model):
username = models.CharField(max_length=20, unique=True)
password = PasswordField(max_length=100)
signup_time = models.DateTimeField(auto_now_add=True)
class Session(AbstractSession):
user = models.ForeignKey(
User, related_name='sessions',
null=True, default=None, on_delete=models.CASCADE
)
我們首先編寫了一個用户模型 User, 其中包含了以下字段
username:用户名字段,需要是不能重複的(unique=True)password:密碼字段,使用的 PasswordField 會自動對輸入的明文密碼進行哈希加密signup_time:註冊時間字段
可以看到除了 User 模型外,我們還編寫了一個用户記錄用户會話和登錄狀態的 Session 模型,繼承自 UtilMeta 提供的模型基類 AbstractSession,我們將通過這個模型實現用户的登錄與鑑權
初始化數據庫
當我們編寫好數據模型後即可使用 Django 提供的遷移命令方便地創建對應的數據表了,由於我們使用的是 SQLite,所以無需提前安裝數據庫軟件,只需要運行以下兩行命令即可完成數據庫的創建
meta makemigrations
meta migrate
當看到以下輸出時即表示你已完成了數據庫的創建
Running migrations:
Applying contenttypes.0001_initial... OK
Applying user.0001_initial... OK
數據庫遷移命令根據 server.py 中的數據庫配置,在項目文件夾中創建了一個名為 db 的 SQLite 數據庫,其中已經完成了 User 和 Session 模型的建表
3. 配置 Session 與用户鑑權
編寫完用户鑑權相關的模型,我們就可以開始開發鑑權相關的邏輯了,我們在 user 文件夾中新建一個 auth.py 文件,編寫 Session 與用户鑑權的配置
from utilmeta.core import auth
from utilmeta.core.auth.session.db import DBSessionSchema, DBSession
from .models import Session, User
USER_ID = '_user_id'
class SessionSchema(DBSessionSchema):
def get_session_data(self):
data = super().get_session_data()
data.update(user_id=self.get(USER_ID))
return data
session_config = DBSession(
session_model=Session,
engine=SessionSchema,
cookie=DBSession.Cookie(
name='sessionid',
age=7 * 24 * 3600,
http_only=True
)
)
user_config = auth.User(
user_model=User,
authentication=session_config,
key=USER_ID,
login_fields=User.username,
password_field=User.password,
)
在這段代碼中,SessionSchema 是處理和存儲 Session 數據的核心引擎,session_config 是聲明 Session 配置的組件,定義了我們剛編寫的 Session 模型以及引擎,並且配置了相應的 Cookie 策略
為了簡化案例,我們選擇了基於數據庫的 Session 實現(DBSession),實際開發中,我們常常使用 Redis 等緩存作為 Session 的存儲實現,或者使用 緩存+數據庫 的方式,這些實現方式 UtilMeta 都支持,你可以在 Session 鑑權文檔 中找到更多的使用方式
另外在代碼中我們也聲明瞭 user_config 用户鑑權配置,其中的參數包括
user_model:指定鑑權的用户模型,就是我上一節中編寫好的 User 模型authentication:指定鑑權策略,我們傳入剛剛定義的session_config,表示着用户鑑權使用 Session 進行key:在 Session 數據中保存當前用户 ID 的名稱,默認是'_user_id'login_fields:能用於登錄的字段,如用户標識名,郵箱等,需要是唯一的password_field:用户的密碼字段,聲明這些可以讓 UtilMeta 自動幫你處理登錄校驗邏輯
4. 編寫用户 API
註冊接口
我們首先來編寫用户的註冊接口,註冊接口應該接收用户名,密碼字段,校驗用户名沒有被佔用後完成註冊,並返回新註冊的用户數據
我們打開 user/api.py 編寫註冊接口
from datetime import datetime
from utilmeta.core import api, orm
from utilmeta.utils import exceptions
from .models import User
from . import auth
class SignupSchema(orm.Schema[User]):
username: str
password: str
class UserSchema(orm.Schema[User]):
id: int
username: str
signup_time: datetime
@auth.session_config.plugin
class UserAPI(api.API):
@api.post
def signup(self, data: SignupSchema = request.Body) -> UserSchema:
if User.objects.filter(username=data.username).exists():
raise exceptions.BadRequest('Username exists')
data.save()
auth.user_config.login_user(
request=self.request,
user=data.get_instance()
)
return UserSchema.init(data.pk)
我們使用 @api 裝飾器定義提供接口服務的 API 函數,其中有 get / post / put / patch / delete 等 HTTP 方法,我們註冊接口使用的是 post 方法,你可以使用裝飾器的第一個參數指定接口的路徑,如果沒有的話(比如例子中),就會自動使用函數的名稱(signup)作為路徑
我們首先聲明瞭註冊接口所接受的數據結構 SignupSchema 作為請求體(request.Body)的類型聲明,這樣 UtilMeta 就會對註冊接口的請求體進行解析並轉化為一個 SignupSchema 實例,不符合要求的請求會被框架自動拒絕並返回 400 響應
註冊接口中的邏輯為
- 檢測請求中的
username是否已被註冊 - 調用
data.save()方法保存數據 - 為當前請求使用
login_user方法登錄新註冊的用户 - 使用
UserSchema.init(data.pk)將新用户的數據初始化為UserSchema實例後返回
UtilMeta 實現了一套高效的聲明式 ORM 查詢體系,我們在聲明 Schema 類時便使用 orm.Schema[User] 綁定了模型,這樣我們就可以通過它的方法來實現數據的增刪改查了,你可以在 數據查詢與 ORM 文檔 中查看它的更多用法
另外我們發現在 UserAPI 類被施加了 @auth.session_config.plugin 這一裝飾器插件,這是 Session 配置應用到 API 上的方式,這個插件能在每次請求結束後對請求所更新的 Session 數據進行保存,並返回對應的 Set-Cookie
登錄登出接口
接下來我們編寫用户的登錄與登出接口
from datetime import datetime
from utilmeta.core import api, orm, request
from utilmeta.utils import exceptions
from .models import User
from . import auth
import utype
class LoginSchema(utype.Schema):
username: str
password: str
@auth.session_config.plugin
class UserAPI(api.API):
@api.post
def signup(self): ...
# new ++++
@api.post
def login(self, data: LoginSchema = request.Body) -> UserSchema:
user = auth.user_config.login(
request=self.request,
ident=data.username,
password=data.password
)
if not user:
raise exceptions.PermissionDenied('Username of password wrong')
return UserSchema.init(user)
@api.post
def logout(self, session: auth.SessionSchema = auth.session_config):
session.flush()
在登錄接口中,我們直接調用了鑑權配置中的 login() 方法來完成登錄,由於我們已經配置好了登錄字段與密碼字段,UtilMeta 可以自動幫我們完成密碼校驗與登錄,如果成功登錄,便返回相應的用户實例
所以當返回為空時,我們便拋出錯誤返回登錄失敗,而成功登錄後,我們調用 UserSchema.init 方法將登錄的用户數據返回給客户端
而對於登出接口,我們只需將當前請求 Session 的數據清空即可,我們使用之前聲明的 session_config 作為 API 函數參數的默認值從而接收當前請求的 Session 對象,然後在函數中調用 session.flush() 清空 Session 數據
用户信息的獲取與更新
當我們瞭解了 Schema Query 的用法後,編寫用户信息的獲取與更新接口就非常簡單了,如下
from datetime import datetime
from utilmeta.core import api, orm, request
from utilmeta.utils import exceptions
from .models import User
from . import auth
import utype
class UserUpdateSchema(orm.Schema[User]):
id: int = orm.Field(no_input=True)
username: str = orm.Field(required=False)
password: str = orm.Field(required=False)
@auth.session_config.plugin
class UserAPI(api.API):
@api.post
def signup(self): ...
@api.post
def login(self): ...
@api.post
def logout(self): ...
# new ++++
def get(self, user: User = auth.user_config) -> UserSchema:
return UserSchema.init(user)
def put(self, data: UserUpdateSchema = request.Body,
user: User = auth.user_config) -> UserSchema:
data.id = user.pk
data.save()
return UserSchema.init(data.pk)
當我們聲明瞭用户鑑權配置後,在任何一個需要用户登錄才能訪問的接口,我們都可以在接口參數中聲明 user: User = auth.user_config 從而拿到當前請求用户的實例,如果請求沒有登錄,則 UtilMeta 會自動處理並返回 401 Unauthorized
在 get 接口中,我們直接將當前的請求用户的數據用 UserSchema 初始化並返回給客户端
在 put 接口中,我們將當前用户的 ID 賦值給接收到 UserUpdateSchema 實例的 id 字段,然後保存並返回更新後的用户數據
由於我們不能允許請求用户任意指定要更新的用户 ID,所以對於請求數據的 id 字段我們使用了 no_input=True 的選項,這其實也是一種常見的權限策略,即一個用户只能更新自己的信息
當你的函數直接使用 get / put / patch / post / delete 等 HTTP 動詞進行命名時,它們就會自動綁定對應的方法,路徑與 API 類的路徑保持一致,這些方法稱為這個 API 類的核心方法
至此我們的 API 就全部開發完成了
整合 API
為了使我們開發的 UserAPI 能夠提供訪問,我們需要把它 掛載 到服務的根 API 上,我們回到 server.py,修改 RootAPI 的聲明
from utilmeta.core.server.backends.django import DjangoSettings
service.use(DjangoSettings(
secret_key='YOUR_SECRET_KEY',
apps=['user']
))
# new +++
service.setup()
from user.api import UserAPI
class RootAPI(api.API):
user: UserAPI
service.mount(RootAPI, route='/api')
我們將開發好的 UserAPI 掛載到了 RootAPI 的 user 屬性,意味着 UserAPI 的路徑被掛載到了 /api/user,其中定義的接口路徑也相應延申,如
GET /api/user:獲取用户信息PUT /api/user:更新用户信息POST /api/user/login:用户登錄POST /api/user/logout:用户登出POST /api/user/signup:用户註冊
這樣的 API 樹級掛載對於組織接口架構和定義樹狀的接口路由非常方便
對於使用 Django 的 API 服務,請在導入任何模型或 API 前加入 service.setup() 完成服務的初始化,這樣 django 才能正確識別所有的數據模型
5. 運行 API
在項目文件夾中使用如下命令即可將 API 服務運行起來
meta run
或者你也可以使用
python server.py
當你看到如下輸出時表示服務已成功啓動
UtilMeta v2.4.1 starting service [demo-user]
...
Starting development server at http://127.0.0.1:8000/
你可以通過調整server.py中的 UtilMeta 服務聲明裏的host和port參數來改變 API 服務監聽的地址
6. 連接 API 管理
UtilMeta 框架內置了一個 API 服務管理系統,你可以方便地連接到你的 API 服務,查看 API 文檔,日誌,監控和管理數據,對於我們開發好的用户登錄註冊接口,只需要在 server.py 中給服務添加以下配置
from utilmeta.ops import Operations
service.use(Operations(
route='ops',
database=Database(
name='operations_db',
engine='sqlite3',
)
))
我們的 Operations 配置將提供觀測與管理功能的 OperationsAPI 掛載到了 ops 路徑,並使用一個 SQLite3 數據庫存儲日誌和監控等運維數據(你也可以在生產時連接 PostgreSQL 等數據庫),
我們重啓項目後可以看到以下輸出
UtilMeta OperationsAPI loaded at http://127.0.0.1:8000/api/ops, connect your APIs at https://ops.utilmeta.com
你可以在一個新的控制枱窗口中進入到 demo-user 文件夾並運行
meta connect
如果你沒有更改服務的端口號和路徑配置的話,你也可以直接點擊 這個鏈接 連接
我們點擊 API 即可看到我們剛開發的用户登錄註冊 API,我們可以選擇註冊 API,並點擊【Debug】按鈕進行測試創建一個用户
註冊成功後你可以點擊左欄的 Data 板塊查看新創建的用户數據
我們可以看到剛剛註冊的用户數據,下方是用户模型的表結構文檔,我們可以選中表記錄進行編輯或刪除,也可以點擊右上角的【+】創建新的實例
你也可以點擊左欄的 Logs 板塊查看註冊接口的調用日誌
點擊日誌展開可以看到詳細的請求和響應數據
你可以注意到數據與日誌板塊的密碼字段都被自動隱藏了起來,因為它們的名稱命中了 Operations 配置的默認 secret_names,更詳細的 API 服務連接與管理配置與功能可以參考 運維與監控管理文檔
由於受限於瀏覽器的規則,Web 端的本地調試無法發送 Cookie 信息,如果你希望更全面地調試開發好的接口,可以參考接下來的部分
7. 調試 API
啓動好 API 服務後我們就可以調試我們的接口了,我們可以使用 UtilMeta 自帶的客户端測試工具方便地調試接口,我們在項目目錄中新建一個 test.py 文件,寫入調試 API 的代碼
from server import service
if __name__ == '__main__':
with service.get_client(live=True) as client:
r1 = client.post('user/signup', data={
'username': 'user1',
'password': '123123'
})
r1.print()
r2 = client.get('user')
r2.print()
其中編寫了用户註冊接口和獲取當前用户接口的調試代碼,當我們啓動服務並運行 test.py 時,我們可以看到的輸出類似
Response [200 OK] "POST /api/user/signup"
application/json (76)
{'username': 'user1', 'id': 1, 'signup_time': '2024-01-29T12:29:33.684594'}
Response [200 OK] "GET /api/user"
application/json (76)
{'username': 'user1', 'id': 1, 'signup_time': '2024-01-29T12:29:33.684594'}
這説明我們的註冊接口和獲取用户的接口開發成功,首先註冊接口返回了正確的結果,然後註冊接口登錄了新註冊的用户,所以之後訪問用户獲取接口也得到了同樣的結果
在with代碼塊中,客户端會記憶響應中Set-Cookie所存儲的 cookies 併發送到接下來的請求中,所以我們可以看到與真實的瀏覽器類似的會話效果
UtilMeta 服務實例的 get_client 方法用於獲取一個服務的客户端實例,你可以直接調用這個實例的 get, post 等方法發起 HTTP 請求,將會得到一個 utilmeta.core.response.Response 響應,這與我們在 API 服務中生成的響應類型一致,其中常用的屬性有
status:響應的狀態碼data:解析後的響應數據,如果是 JSON 響應體,則會得到一個dict或list類型的數據headers:響應頭request:響應對應的請求對象,有請求的方法,路徑等參數信息
get_client方法中的live參數如果沒有開啓,則是直接調用對應的接口函數進行調試,無需啓動服務
所以你也可以使用這個客户端編寫單元測試,比如
from server import service
def test_signup():
with service.get_client(live=True) as client:
r1 = client.post('user/signup', data={
'username': 'user1',
'password': '123123'
})
assert r1.status == 200
assert isinstance(r1.data, dict)
assert r1.data.get('username') == 'user1'
我們還可以測試登錄,登出與更新接口,比如在登出後 cookies 應該被清空,之後獲取當前用户也應該返回空,最後完整的調試代碼與對應的輸出如下
from server import service
if __name__ == '__main__':
with service.get_client(live=True) as client:
r1 = client.post('user/signup', data={
'username': 'user1',
'password': '123123'
})
r1.print()
# Response [200 OK] "POST /api/user/signup"
# application/json (75)
# {'username': 'user1', 'id': 1, 'signup_time': '2024-01-29T13:29:03.336134'}
r2 = client.get('user')
r2.print()
# Response [200 OK] "GET /api/user"
# application/json (75)
# {'username': 'user1', 'id': 1, 'signup_time': '2024-01-29T13:29:03.336134'}
r3 = client.post('user/logout')
r3.print()
# Response [200 OK] "POST /api/user/logout"
# text/html (0)
r4 = client.get('user')
r4.print()
# Response [401 Unauthorized] "GET /api/user"
# text/html (0)
r5 = client.post('user/login', data={
'username': 'user1',
'password': '123123'
})
# Response [200 OK] "POST /api/user/login"
# application/json (75)
# {'username': 'user1', 'id': 1, 'signup_time': '2024-01-29T13:29:03.336134'}
r5.print()
r6 = client.get('user')
r6.print()
# Response [200 OK] "GET /api/user"
# application/json (75)
# {'username': 'user1', 'id': 1, 'signup_time': '2024-01-29T13:29:03.336134'}
r7 = client.put('user', data={
'username': 'user-updated',
'password': '123456'
})
r7.print()
# Response [200 OK] "PUT /api/user"
# application/json (82)
# {'username': 'user-updated', 'id': 1, 'signup_time': '2024-01-29T13:44:30.095711'}
源碼與資料
- 案例源碼:Github
- 框架首頁:UtilMeta
我同時也是 UtilMeta 框架的作者,如果你有什麼問題或者想加入 UtilMeta 開發者羣也歡迎聯繫我,我的全網 ID 和微信號都是 voidZXL