1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
| from typing import Union
from flask.views import MethodView
from flask import request, g
from apps.api_v1 import api_v1
from apps.forms.user import SignUpForm
from apps.utils.response import HttpResponse
from apps.extensions import db
from apps.models.user import User
class UserAPI(MethodView):
USER_CACHR_PREFIX = "USERID::"
def extract_user_info(self, user: User) -> dict:
return {
"username": user.username,
"id": user.id,
"age": user.age,
"mobile": user.mobile
}
def get_user_info(self, user: Union[User,int]) -> dict:
if isinstance(user, User):
result = self.extract_user_info(user)
self.add_cache(result)
return result
elif isinstance(user, int):
result = g.redis.hgetall(f'{self.USER_CACHR_PREFIX}{user}')
if result:
result["cache"] = True
return result
else:
user = User.query.get(user)
if not user:
return {}
else:
result = self.extract_user_info(user)
self.add_cache(result)
return result
def add_cache(self, user_info: dict):
g.redis.hmset(f'{self.USER_CACHR_PREFIX}{user_info["id"]}', user_info)
def delete_cache(self, user_id: int):
g.redis.delete(f'{self.USER_CACHR_PREFIX}{user_id}')
def user_not_found(self, user_id: int) -> str:
return f"user not found: {user_id}"
def get(self, user_id: int):
if user_id is None:
users = User.query.all()
data = [self.get_user_info(item) for item in users]
return HttpResponse.success(data={"data": data})
else:
data = self.get_user_info(user_id)
if not data:
return HttpResponse.params_error(message=self.user_not_found(user_id))
else:
return HttpResponse.success(data=data)
def post(self):
form = SignUpForm.from_json(request.json)
if form.validate():
user = User(
username=form.username.data,
password=form.password.data,
mobile=form.mobile.data,
age=form.info.age.data,
birth=form.info.birth.data
)
with db.auto_commit():
db.session.add(user)
self.add_cache(self.extract_user_info(user))
return HttpResponse.success(data=self.get_user_info(user))
else:
message = f"add user fail:{form.errors}"
return HttpResponse.params_error(message=message)
def delete(self, user_id: int):
user = User.query.get(user_id)
if not user:
return HttpResponse.params_error(message=self.user_not_found(user_id))
else:
with db.auto_commit():
db.session.delete(user)
self.delete_cache(user.id)
return HttpResponse.success(message=f'delete user success: {user_id}')
def put(self, user_id: int):
form = SignUpForm.from_json(request.json)
if form.validate():
user = User.query.get(user_id)
if not user:
return HttpResponse.params_error(message=self.user_not_found(user_id))
with db.auto_commit():
user.username=form.username.data
user.password=form.password.data
user.mobile=form.mobile.data
user.age=form.info.age.data
user.birth=form.info.birth.data
self.delete_cache(user.id)
return HttpResponse.success(data=self.get_user_info(user))
else:
message = f"update user fail:{form.errors}"
return HttpResponse.params_error(message=message)
user_view = UserAPI.as_view('user_api')
api_v1.add_url_rule('/users/', defaults={'user_id': None}, view_func=user_view, methods=['GET',])
api_v1.add_url_rule('/users/', view_func=user_view, methods=['POST',])
api_v1.add_url_rule('/users/<int:user_id>', view_func=user_view, methods=['GET', 'PUT', 'DELETE'])
|