在開(kāi)發(fā)大多數(shù)應(yīng)用時(shí)，用戶系統(tǒng)都是必不可少的部分，而我們總是需要開(kāi)發(fā)圍繞用戶的登錄，注冊(cè)，獲取，更新等接口。在這篇文章將帶你用一百多行代碼簡(jiǎn)潔地實(shí)現(xiàn)一套這樣的用戶鑒權(quán)與 RESTful 接口，并使用 Session 來(lái)處理用戶的登錄登出

我們將使用 UtilMeta 框架 完成接口開(kāi)發(fā)，這是一個(gè)開(kāi)源的 Python 后端元框架，同時(shí)支持接入與適配 Django, Flask, FastAPI 等主流 Python 框架，并且能簡(jiǎn)潔高效地開(kāi)發(fā) RESTful 接口

0. 安裝框架

使用如下命令即可安裝 UtilMeta 框架

pip install utilmeta

UtilMeta 框架需要 Python 版本 >= 3.8

1. 創(chuàng)建項(xiàng)目

我們使用如下命令來(lái)創(chuàng)建一個(gè)新項(xiàng)目

meta setup demo-user

我們的項(xiàng)目將會(huì)使用 Django 作為底層框架，所以在提示選擇 backend 的時(shí)候我們可以輸入 django

項(xiàng)目創(chuàng)建好后，我們需要先對(duì)服務(wù)的數(shù)據(jù)庫(kù)連接進(jìn)行配置，打開(kāi) 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 的配置信息與數(shù)據(jù)庫(kù)連接的配置
由于 Django 使用 app (應(yīng)用) 的方式來(lái)管理數(shù)據(jù)模型，接下來(lái)我們使用如下的命令來(lái)創(chuàng)建一個(gè)名為 user 的 app

cd demo-user
meta add user

可以看到在我們的項(xiàng)目文件夾中新創(chuàng)建出了一個(gè) user 文件夾，其中包括

/user
    /migrations
    api.py
    models.py
    schema.py

其中 migrations 文件夾是 Django 用來(lái)處理數(shù)據(jù)庫(kù)遷移文件的，models.py 是我們編寫(xiě)數(shù)據(jù)模型的地方

應(yīng)用創(chuàng)建完成后，我們將 server.py 的 Django 設(shè)置中插入一行代碼來(lái)注入新創(chuàng)建的 user app

service.use(DjangoSettings(
    secret_key='YOUR_SECRET_KEY',
    apps=['user']     # new
))

至此我們完成了項(xiàng)目的配置和初始化

2. 編寫(xiě)用戶模型

用戶的登錄注冊(cè) API 當(dāng)然是圍繞 “用戶” 進(jìn)行的了，在開(kāi)發(fā) API 之前，我們需要先編寫(xiě)好用戶的數(shù)據(jù)模型，我們打開(kāi) user/models.py，編寫(xiě)

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
    )

我們首先編寫(xiě)了一個(gè)用戶模型 User, 其中包含了以下字段

• username：用戶名字段，需要是不能重復(fù)的（unique=True）
• password：密碼字段，使用的 PasswordField 會(huì)自動(dòng)對(duì)輸入的明文密碼進(jìn)行哈希加密
• signup_time：注冊(cè)時(shí)間字段

可以看到除了 User 模型外，我們還編寫(xiě)了一個(gè)用戶記錄用戶會(huì)話和登錄狀態(tài)的 Session 模型，繼承自 UtilMeta 提供的模型基類 AbstractSession，我們將通過(guò)這個(gè)模型實(shí)現(xiàn)用戶的登錄與鑒權(quán)

初始化數(shù)據(jù)庫(kù)

當(dāng)我們編寫(xiě)好數(shù)據(jù)模型后即可使用 Django 提供的遷移命令方便地創(chuàng)建對(duì)應(yīng)的數(shù)據(jù)表了，由于我們使用的是 SQLite，所以無(wú)需提前安裝數(shù)據(jù)庫(kù)軟件，只需要運(yùn)行以下兩行命令即可完成數(shù)據(jù)庫(kù)的創(chuàng)建

meta makemigrations
meta migrate

當(dāng)看到以下輸出時(shí)即表示你已完成了數(shù)據(jù)庫(kù)的創(chuàng)建

Running migrations:
  Applying contenttypes.0001_initial... OK
  Applying user.0001_initial... OK

數(shù)據(jù)庫(kù)遷移命令根據(jù) server.py 中的數(shù)據(jù)庫(kù)配置，在項(xiàng)目文件夾中創(chuàng)建了一個(gè)名為 db 的 SQLite 數(shù)據(jù)庫(kù)，其中已經(jīng)完成了 User 和 Session 模型的建表

3. 配置 Session 與用戶鑒權(quán)

編寫(xiě)完用戶鑒權(quán)相關(guān)的模型，我們就可以開(kāi)始開(kāi)發(fā)鑒權(quán)相關(guān)的邏輯了，我們?cè)?user 文件夾中新建一個(gè) auth.py 文件，編寫(xiě) Session 與用戶鑒權(quán)的配置

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 是處理和存儲(chǔ) Session 數(shù)據(jù)的核心引擎，session_config 是聲明 Session 配置的組件，定義了我們剛編寫(xiě)的 Session 模型以及引擎，并且配置了相應(yīng)的 Cookie 策略

為了簡(jiǎn)化案例，我們選擇了基于數(shù)據(jù)庫(kù)的 Session 實(shí)現(xiàn)（DBSession），實(shí)際開(kāi)發(fā)中，我們常常使用 Redis 等緩存作為 Session 的存儲(chǔ)實(shí)現(xiàn)，或者使用 緩存+數(shù)據(jù)庫(kù) 的方式，這些實(shí)現(xiàn)方式 UtilMeta 都支持，你可以在 Session 鑒權(quán)文檔 中找到更多的使用方式

另外在代碼中我們也聲明了 user_config 用戶鑒權(quán)配置，其中的參數(shù)包括

• user_model：指定鑒權(quán)的用戶模型，就是我上一節(jié)中編寫(xiě)好的 User 模型
• authentication：指定鑒權(quán)策略，我們傳入剛剛定義的 session_config，表示著用戶鑒權(quán)使用 Session 進(jìn)行
• key：在 Session 數(shù)據(jù)中保存當(dāng)前用戶 ID 的名稱，默認(rèn)是 '_user_id'
• login_fields：能用于登錄的字段，如用戶標(biāo)識(shí)名，郵箱等，需要是唯一的
• password_field：用戶的密碼字段，聲明這些可以讓 UtilMeta 自動(dòng)幫你處理登錄校驗(yàn)邏輯

4. 編寫(xiě)用戶 API

注冊(cè)接口

我們首先來(lái)編寫(xiě)用戶的注冊(cè)接口，注冊(cè)接口應(yīng)該接收用戶名，密碼字段，校驗(yàn)用戶名沒(méi)有被占用后完成注冊(cè)，并返回新注冊(cè)的用戶數(shù)據(jù)

我們打開(kāi) user/api.py 編寫(xiě)注冊(cè)接口

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 裝飾器定義提供接口服務(wù)的 API 函數(shù)，其中有 get / post / put / patch / delete 等 HTTP 方法，我們注冊(cè)接口使用的是 post 方法，你可以使用裝飾器的第一個(gè)參數(shù)指定接口的路徑，如果沒(méi)有的話（比如例子中），就會(huì)自動(dòng)使用函數(shù)的名稱（signup）作為路徑

我們首先聲明了注冊(cè)接口所接受的數(shù)據(jù)結(jié)構(gòu) SignupSchema 作為請(qǐng)求體（request.Body）的類型聲明，這樣 UtilMeta 就會(huì)對(duì)注冊(cè)接口的請(qǐng)求體進(jìn)行解析并轉(zhuǎn)化為一個(gè) SignupSchema 實(shí)例，不符合要求的請(qǐng)求會(huì)被框架自動(dòng)拒絕并返回 400 響應(yīng)

注冊(cè)接口中的邏輯為

1. 檢測(cè)請(qǐng)求中的 username 是否已被注冊(cè)
2. 調(diào)用 data.save() 方法保存數(shù)據(jù)
3. 為當(dāng)前請(qǐng)求使用 login_user 方法登錄新注冊(cè)的用戶
4. 使用 UserSchema.init(data.pk) 將新用戶的數(shù)據(jù)初始化為 UserSchema 實(shí)例后返回

UtilMeta 實(shí)現(xiàn)了一套高效的聲明式 ORM 查詢體系，我們?cè)诼暶?Schema 類時(shí)便使用 orm.Schema[User] 綁定了模型，這樣我們就可以通過(guò)它的方法來(lái)實(shí)現(xiàn)數(shù)據(jù)的增刪改查了，你可以在 數(shù)據(jù)查詢與 ORM 文檔 中查看它的更多用法

另外我們發(fā)現(xiàn)在 UserAPI 類被施加了 @auth.session_config.plugin 這一裝飾器插件，這是 Session 配置應(yīng)用到 API 上的方式，這個(gè)插件能在每次請(qǐng)求結(jié)束后對(duì)請(qǐng)求所更新的 Session 數(shù)據(jù)進(jìn)行保存，并返回對(duì)應(yīng)的 Set-Cookie

登錄登出接口

接下來(lái)我們編寫(xiě)用戶的登錄與登出接口

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()

在登錄接口中，我們直接調(diào)用了鑒權(quán)配置中的 login() 方法來(lái)完成登錄，由于我們已經(jīng)配置好了登錄字段與密碼字段，UtilMeta 可以自動(dòng)幫我們完成密碼校驗(yàn)與登錄，如果成功登錄，便返回相應(yīng)的用戶實(shí)例

所以當(dāng)返回為空時(shí)，我們便拋出錯(cuò)誤返回登錄失敗，而成功登錄后，我們調(diào)用 UserSchema.init 方法將登錄的用戶數(shù)據(jù)返回給客戶端

而對(duì)于登出接口，我們只需將當(dāng)前請(qǐng)求 Session 的數(shù)據(jù)清空即可，我們使用之前聲明的 session_config 作為 API 函數(shù)參數(shù)的默認(rèn)值從而接收當(dāng)前請(qǐng)求的 Session 對(duì)象，然后在函數(shù)中調(diào)用 session.flush() 清空 Session 數(shù)據(jù)

用戶信息的獲取與更新

當(dāng)我們了解了 Schema Query 的用法后，編寫(xiě)用戶信息的獲取與更新接口就非常簡(jiǎn)單了，如下

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)

當(dāng)我們聲明了用戶鑒權(quán)配置后，在任何一個(gè)需要用戶登錄才能訪問(wèn)的接口，我們都可以在接口參數(shù)中聲明 user: User = auth.user_config 從而拿到當(dāng)前請(qǐng)求用戶的實(shí)例，如果請(qǐng)求沒(méi)有登錄，則 UtilMeta 會(huì)自動(dòng)處理并返回 401 Unauthorized

在 get 接口中，我們直接將當(dāng)前的請(qǐng)求用戶的數(shù)據(jù)用 UserSchema 初始化并返回給客戶端
在 put 接口中，我們將當(dāng)前用戶的 ID 賦值給接收到 UserUpdateSchema 實(shí)例的 id 字段，然后保存并返回更新后的用戶數(shù)據(jù)

由于我們不能允許請(qǐng)求用戶任意指定要更新的用戶 ID，所以對(duì)于請(qǐng)求數(shù)據(jù)的 id 字段我們使用了 no_input=True 的選項(xiàng)，這其實(shí)也是一種常見(jiàn)的權(quán)限策略，即一個(gè)用戶只能更新自己的信息

當(dāng)你的函數(shù)直接使用 get / put / patch / post / delete 等 HTTP 動(dòng)詞進(jìn)行命名時(shí)，它們就會(huì)自動(dòng)綁定對(duì)應(yīng)的方法，路徑與 API 類的路徑保持一致，這些方法稱為這個(gè) API 類的核心方法

至此我們的 API 就全部開(kāi)發(fā)完成了

整合 API

為了使我們開(kāi)發(fā)的 UserAPI 能夠提供訪問(wèn)，我們需要把它 掛載 到服務(wù)的根 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')

我們將開(kāi)發(fā)好的 UserAPI 掛載到了 RootAPI 的 user 屬性，意味著 UserAPI 的路徑被掛載到了 /api/user，其中定義的接口路徑也相應(yīng)延申，如

• GET /api/user：獲取用戶信息
• PUT /api/user：更新用戶信息
• POST /api/user/login：用戶登錄
• POST /api/user/logout：用戶登出
• POST /api/user/signup：用戶注冊(cè)

這樣的 API 樹(shù)級(jí)掛載對(duì)于組織接口架構(gòu)和定義樹(shù)狀的接口路由非常方便

對(duì)于使用 Django 的 API 服務(wù)，請(qǐng)?jiān)趯?dǎo)入任何模型或 API 前加入 service.setup() 完成服務(wù)的初始化，這樣 django 才能正確識(shí)別所有的數(shù)據(jù)模型

5. 運(yùn)行 API

在項(xiàng)目文件夾中使用如下命令即可將 API 服務(wù)運(yùn)行起來(lái)

meta run

或者你也可以使用

python server.py

當(dāng)你看到如下輸出時(shí)表示服務(wù)已成功啟動(dòng)

UtilMeta v2.4.1 starting service [demo-user]
...
Starting development server at http://127.0.0.1:8000/

你可以通過(guò)調(diào)整 server.py 中的 UtilMeta 服務(wù)聲明里的 host 和 port 參數(shù)來(lái)改變 API 服務(wù)監(jiān)聽(tīng)的地址

6. 調(diào)試 API

啟動(dòng)好 API 服務(wù)后我們就可以調(diào)試我們的接口了，我們可以使用 UtilMeta 自帶的客戶端測(cè)試工具方便地調(diào)試接口，我們?cè)陧?xiàng)目目錄中新建一個(gè) test.py 文件，寫(xiě)入調(diào)試 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()

其中編寫(xiě)了用戶注冊(cè)接口和獲取當(dāng)前用戶接口的調(diào)試代碼，當(dāng)我們啟動(dòng)服務(wù)并運(yùn)行 test.py 時(shí)，我們可以看到的輸出類似

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'}

這說(shuō)明我們的注冊(cè)接口和獲取用戶的接口開(kāi)發(fā)成功，首先注冊(cè)接口返回了正確的結(jié)果，然后注冊(cè)接口登錄了新注冊(cè)的用戶，所以之后訪問(wèn)用戶獲取接口也得到了同樣的結(jié)果

在 with 代碼塊中，客戶端會(huì)記憶響應(yīng)中 Set-Cookie 所存儲(chǔ)的 cookies 并發(fā)送到接下來(lái)的請(qǐng)求中，所以我們可以看到與真實(shí)的瀏覽器類似的會(huì)話效果

UtilMeta 服務(wù)實(shí)例的 get_client 方法用于獲取一個(gè)服務(wù)的客戶端實(shí)例，你可以直接調(diào)用這個(gè)實(shí)例的 get, post 等方法發(fā)起 HTTP 請(qǐng)求，將會(huì)得到一個(gè) utilmeta.core.response.Response 響應(yīng)，這與我們?cè)?API 服務(wù)中生成的響應(yīng)類型一致，其中常用的屬性有

• status：響應(yīng)的狀態(tài)碼
• data：解析后的響應(yīng)數(shù)據(jù)，如果是 JSON 響應(yīng)體，則會(huì)得到一個(gè) dict 或 list 類型的數(shù)據(jù)
• headers：響應(yīng)頭
• request：響應(yīng)對(duì)應(yīng)的請(qǐng)求對(duì)象，有請(qǐng)求的方法，路徑等參數(shù)信息

get_client 方法中的 live 參數(shù)如果沒(méi)有開(kāi)啟，則是直接調(diào)用對(duì)應(yīng)的接口函數(shù)進(jìn)行調(diào)試，無(wú)需啟動(dòng)服務(wù)

所以你也可以使用這個(gè)客戶端編寫(xiě)單元測(cè)試，比如

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'

我們還可以測(cè)試登錄，登出與更新接口，比如在登出后 cookies 應(yīng)該被清空，之后獲取當(dāng)前用戶也應(yīng)該返回空，最后完整的調(diào)試代碼與對(duì)應(yīng)的輸出如下

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
• 框架首頁(yè)：UtilMeta

我同時(shí)也是 UtilMeta 框架的作者，如果你有什么問(wèn)題也歡迎聯(lián)系我，我的全網(wǎng) ID 和微信號(hào)都是 voidZXL文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-825159.html

到了這里，關(guān)于100 行代碼實(shí)現(xiàn)用戶登錄注冊(cè)與 RESTful 接口 - 手把手教程附 Python 源碼的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！