100 行代码实现用户登录注册与 RESTful 接口 - 手把手教程附 Python 源码 ...

在开发大多数应用时，用户系统都是必不可少的部分，而我们总是需要开发围绕用户的登录，注册，获取，更新等接口。在这篇文章将带你用一百多行代码简洁地实现一套这样的用户鉴权与 RESTful 接口，并使用 Session 来处理用户的登录登出
我们将使用 UtilMeta 框架 完成接口开发，这是一个开源的 Python 后端元框架，同时支持接入与适配 Django, Flask, FastAPI 等主流 Python 框架，并且能简洁高效地开发 RESTful 接口
0. 安装框架

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

1. pip install utilmeta

复制代码

UtilMeta 框架需要 Python 版本 >= 3.8

1. 创建项目

我们使用如下命令来创建一个新项目

1. meta setup demo-user

复制代码

我们的项目将会使用 Django 作为底层框架，所以在提示选择 backend 的时候我们可以输入 django
项目创建好后，我们需要先对服务的数据库连接进行配置，打开 server.py，在 service 的声明下面插入以下代码

1. service = UtilMeta(
2.     __name__,
3.     name='demo-user',
4.     backend=django,
5. )
7. # new +++++
8. from utilmeta.core.server.backends.django import DjangoSettings
9. from utilmeta.core.orm import DatabaseConnections, Database
11. service.use(DjangoSettings(
12.     secret_key='YOUR_SECRET_KEY',
13. ))
15. service.use(DatabaseConnections({
16.     'default': Database(
17.         name='db',
18.         engine='sqlite3',
19.     )
20. }))

复制代码

在插入的代码中，我们声明了 Django 的配置信息与数据库连接的配置
由于 Django 使用 app (应用) 的方式来管理数据模型，接下来我们使用如下的命令来创建一个名为 user 的 app

1. cd demo-user
2. meta add user

复制代码

可以看到在我们的项目文件夹中新创建出了一个 user 文件夹，其中包括

1. /user
2.     /migrations
3.     api.py
4.     models.py
5.     schema.py

复制代码

其中 migrations 文件夹是 Django 用来处理数据库迁移文件的，models.py 是我们编写数据模型的地方
应用创建完成后，我们将 server.py 的 Django 设置中插入一行代码来注入新创建的 user app

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

复制代码

至此我们完成了项目的配置和初始化
2. 编写用户模型

用户的登录注册 API 当然是围绕 “用户” 进行的了，在开发 API 之前，我们需要先编写好用户的数据模型，我们打开 user/models.py，编写

1. from django.db import models
2. from utilmeta.core.orm.backends.django.models import AbstractSession, PasswordField
4. class User(models.Model):
5.     username = models.CharField(max_length=20, unique=True)
6.     password = PasswordField(max_length=100)
7.     signup_time = models.DateTimeField(auto_now_add=True)
9. class Session(AbstractSession):
10.     user = models.ForeignKey(
11.         User, related_name='sessions',
12.         null=True, default=None, on_delete=models.CASCADE
13.     )

复制代码

我们首先编写了一个用户模型 User, 其中包含了以下字段

• username：用户名字段，需要是不能重复的（unique=True）
  
• password：密码字段，使用的 PasswordField 会自动对输入的明文密码进行哈希加密
  
• signup_time：注册时间字段
  

可以看到除了 User 模型外，我们还编写了一个用户记录用户会话和登录状态的 Session 模型，继承自 UtilMeta 提供的模型基类 AbstractSession，我们将通过这个模型实现用户的登录与鉴权
初始化数据库

当我们编写好数据模型后即可使用 Django 提供的迁移命令方便地创建对应的数据表了，由于我们使用的是 SQLite，所以无需提前安装数据库软件，只需要运行以下两行命令即可完成数据库的创建

1. meta makemigrations
2. meta migrate

复制代码

当看到以下输出时即表示你已完成了数据库的创建

1. Running migrations:
2.   Applying contenttypes.0001_initial... OK
3.   Applying user.0001_initial... OK

复制代码

数据库迁移命令根据 server.py 中的数据库配置，在项目文件夹中创建了一个名为 db 的 SQLite 数据库，其中已经完成了 User 和 Session 模型的建表
3. 配置 Session 与用户鉴权

编写完用户鉴权相关的模型，我们就可以开始开发鉴权相关的逻辑了，我们在 user 文件夹中新建一个 auth.py 文件，编写 Session 与用户鉴权的配置

1. from utilmeta.core import auth
2. from utilmeta.core.auth.session.db import DBSessionSchema, DBSession
3. from .models import Session, User
5. USER_ID = '_user_id'
7. class SessionSchema(DBSessionSchema):
8.     def get_session_data(self):
9.         data = super().get_session_data()
10.         data.update(user_id=self.get(USER_ID))
11.         return data
13. session_config = DBSession(
14.     session_model=Session,
15.     engine=SessionSchema,
16.     cookie=DBSession.Cookie(
17.         name='sessionid',
18.         age=7 * 24 * 3600,
19.         http_only=True
20.     )
21. )
23. user_config = auth.User(
24.     user_model=User,
25.     authentication=session_config,
26.     key=USER_ID,
27.     login_fields=User.username,
28.     password_field=User.password,
29. )

复制代码

在这段代码中，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 编写注册接口

1. from datetime import datetime
2. from utilmeta.core import api, orm
3. from utilmeta.utils import exceptions
4. from .models import User
5. from . import auth
7. class SignupSchema(orm.Schema[User]):
8.     username: str
9.     password: str
10.    
11. class UserSchema(orm.Schema[User]):
12.     id: int
13.     username: str
14.     signup_time: datetime
16. @auth.session_config.plugin
17. class UserAPI(api.API):
18.     @api.post
19.     def signup(self, data: SignupSchema = request.Body) -> UserSchema:
20.         if User.objects.filter(username=data.username).exists():
21.             raise exceptions.BadRequest('Username exists')
22.         data.save()
23.         auth.user_config.login_user(
24.             request=self.request,
25.             user=data.get_instance()
26.         )
27.         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
登录登出接口

接下来我们编写用户的登录与登出接口

1. from datetime import datetime
2. from utilmeta.core import api, orm, request
3. from utilmeta.utils import exceptions
4. from .models import User
5. from . import auth
6. import utype
8. class LoginSchema(utype.Schema):
9.     username: str
10.     password: str
12. @auth.session_config.plugin
13. class UserAPI(api.API):
14.     @api.post
15.     def signup(self): ...
17.     # new ++++
18.     @api.post
19.     def login(self, data: LoginSchema = request.Body) -> UserSchema:
20.         user = auth.user_config.login(
21.             request=self.request,
22.             ident=data.username,
23.             password=data.password
24.         )
25.         if not user:
26.             raise exceptions.PermissionDenied('Username of password wrong')
27.         return UserSchema.init(user)
29.     @api.post
30.     def logout(self, session: auth.SessionSchema = auth.session_config):
31.         session.flush()

复制代码

在登录接口中，我们直接调用了鉴权配置中的 login() 方法来完成登录，由于我们已经配置好了登录字段与密码字段，UtilMeta 可以自动帮我们完成密码校验与登录，如果成功登录，便返回相应的用户实例
所以当返回为空时，我们便抛出错误返回登录失败，而成功登录后，我们调用 UserSchema.init 方法将登录的用户数据返回给客户端
而对于登出接口，我们只需将当前请求 Session 的数据清空即可，我们使用之前声明的 session_config 作为 API 函数参数的默认值从而接收当前请求的 Session 对象，然后在函数中调用 session.flush() 清空 Session 数据
用户信息的获取与更新

当我们了解了 Schema Query 的用法后，编写用户信息的获取与更新接口就非常简单了，如下

1. from datetime import datetime
2. from utilmeta.core import api, orm, request
3. from utilmeta.utils import exceptions
4. from .models import User
5. from . import auth
6. import utype
8. class UserUpdateSchema(orm.Schema[User]):
9.     id: int = orm.Field(no_input=True)
10.     username: str = orm.Field(required=False)
11.     password: str = orm.Field(required=False)
13. @auth.session_config.plugin
14. class UserAPI(api.API):
15.     @api.post
16.     def signup(self): ...
17.     @api.post
18.     def login(self): ...
19.     @api.post
20.     def logout(self): ...
22.     # new ++++
23.     def get(self, user: User = auth.user_config) -> UserSchema:
24.         return UserSchema.init(user)
26.     def put(self, data: UserUpdateSchema = request.Body,
27.             user: User = auth.user_config) -> UserSchema:
28.         data.id = user.pk
29.         data.save()
30.         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 的声明

1. from utilmeta.core.server.backends.django import DjangoSettings
3. service.use(DjangoSettings(
4.     secret_key='YOUR_SECRET_KEY',
5.     apps=['user']
6. ))
8. # new +++
9. service.setup()
11. from user.api import UserAPI
13. class RootAPI(api.API):
14.     user: UserAPI
15.        
16. 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 服务运行起来

1. meta run

复制代码

或者你也可以使用

1. python server.py

复制代码

当你看到如下输出时表示服务已成功启动

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

复制代码

你可以通过调整 server.py 中的 UtilMeta 服务声明里的 host 和 port 参数来改变 API 服务监听的地址

6. 调试 API

启动好 API 服务后我们就可以调试我们的接口了，我们可以使用 UtilMeta 自带的客户端测试工具方便地调试接口，我们在项目目录中新建一个 test.py 文件，写入调试 API 的代码

1. from server import service
3. if __name__ == '__main__':
4.     with service.get_client(live=True) as client:
5.         r1 = client.post('user/signup', data={
6.             'username': 'user1',
7.             'password': '123123'
8.         })
9.         r1.print()
10.         r2 = client.get('user')
11.         r2.print()

复制代码

其中编写了用户注册接口和获取当前用户接口的调试代码，当我们启动服务并运行 test.py 时，我们可以看到的输出类似

1. Response [200 OK] "POST /api/user/signup"
2. application/json (76)
3. {'username': 'user1', 'id': 1, 'signup_time': '2024-01-29T12:29:33.684594'}
5. Response [200 OK] "GET /api/user"
6. application/json (76)
7. {'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 参数如果没有开启，则是直接调用对应的接口函数进行调试，无需启动服务

所以你也可以使用这个客户端编写单元测试，比如

1. from server import service
3. def test_signup():
4.     with service.get_client(live=True) as client:
5.         r1 = client.post('user/signup', data={
6.             'username': 'user1',
7.             'password': '123123'
8.         })
9.         assert r1.status == 200
10.         assert isinstance(r1.data, dict)
11.         assert r1.data.get('username') == 'user1'

复制代码

我们还可以测试登录，登出与更新接口，比如在登出后 cookies 应该被清空，之后获取当前用户也应该返回空，最后完整的调试代码与对应的输出如下

1. from server import service
3. if __name__ == '__main__':
4.     with service.get_client(live=True) as client:
5.         r1 = client.post('user/signup', data={
6.             'username': 'user1',
7.             'password': '123123'
8.         })
9.         r1.print()
10.         # Response [200 OK] "POST /api/user/signup"
11.         # application/json (75)
12.         # {'username': 'user1', 'id': 1, 'signup_time': '2024-01-29T13:29:03.336134'}
13.         r2 = client.get('user')
14.         r2.print()
15.         # Response [200 OK] "GET /api/user"
16.         # application/json (75)
17.         # {'username': 'user1', 'id': 1, 'signup_time': '2024-01-29T13:29:03.336134'}
18.         r3 = client.post('user/logout')
19.         r3.print()
20.         # Response [200 OK] "POST /api/user/logout"
21.         # text/html (0)
22.         r4 = client.get('user')
23.         r4.print()
24.         # Response [401 Unauthorized] "GET /api/user"
25.         # text/html (0)
26.         r5 = client.post('user/login', data={
27.             'username': 'user1',
28.             'password': '123123'
29.         })
30.         # Response [200 OK] "POST /api/user/login"
31.         # application/json (75)
32.         # {'username': 'user1', 'id': 1, 'signup_time': '2024-01-29T13:29:03.336134'}
33.         r5.print()
34.         r6 = client.get('user')
35.         r6.print()
36.         # Response [200 OK] "GET /api/user"
37.         # application/json (75)
38.         # {'username': 'user1', 'id': 1, 'signup_time': '2024-01-29T13:29:03.336134'}
39.         r7 = client.put('user', data={
40.             'username': 'user-updated',
41.             'password': '123456'
42.         })
43.         r7.print()
44.         # Response [200 OK] "PUT /api/user"
45.         # application/json (82)
46.         # {'username': 'user-updated', 'id': 1, 'signup_time': '2024-01-29T13:44:30.095711'}

复制代码

源码与资料

• 案例源码：Github
  
• 框架首页：UtilMeta
  

我同时也是 UtilMeta 框架的作者，如果你有什么问题也欢迎联系我，我的全网 ID 和微信号都是 voidZXL

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！