auth_log_query.py
4.3 KB
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
# coding=utf-8
# author: qianyingz
# createtime: 2022/03/09
# email: qianyingz@chinadci.com
from tokenize import String
from numpy import number
from sqlalchemy import or_
from .models import *
from app.util.component.ApiTemplate import ApiTemplate
import json
class Api(ApiTemplate):
api_name = "登录日志"
def para_check(self):
pass
def process(self):
# 返回结果
res = {}
res["result"] = False
try:
# 业务逻辑
page_index = int(self.para.get("page_index", "0"))
page_size = int(self.para.get("page_size", "1000"))
# name = self.para.get("name")
order_by = self.para.get("order_by")
filter_param = self.para.get("filter_param", "")
filter_param_json = {} if filter_param == "" else json.loads(
filter_param)
kw = filter_param_json["kw"] if "kw" in filter_param_json else None
type = filter_param_json["type"] if "type" in filter_param_json else None
log_query = OAuthLog.query
if kw != None:
if isinstance(kw, str):
filter_username_query = log_query.filter(OAuthLog.username.like(
"%" + kw + "%"))
filter_display_query = log_query.filter(
OAuthLog.displayname.like("%" + kw + "%"))
log_query = filter_username_query.union(
filter_display_query)
if type != None:
if isinstance(type, (str,int)):
log_query = log_query.filter(OAuthLog.operate_type == type)
elif isinstance(type, list):
log_query = log_query.filter(OAuthLog.operate_type.in_(type))
log_query = log_query.order_by(OAuthLog.create_time.desc())
count = log_query.count()
logs = log_query.limit(page_size).offset(
page_index*page_size).all()
res["data"] = {"count": count,
"list": list(map(lambda t:
{"id": t.id, "username": t.username, "ip": t.ip,
"message": t.message, "create_time": t.create_time.strftime("%Y-%m-%d %H:%M:%S"),
"operate_type": t.operate_type,
"auth_type": t.auth_type,
"displayname": t.displayname}, logs))}
# # 获取集合
# userLinq = User.query.order_by(User.id.desc())
# if name:
# userLinq = userLinq.filter(
# User.username.like("%" + name + "%"))
# tmp_count = userLinq.count()
# tmp_list = userLinq.limit(page_size).offset(
# page_index * page_size).all()
# res["data"] = {
# "count": tmp_count,
# "list": list(map(lambda t:
# {"guid": t.id, "username": t.username,
# "role": t.role, "display_name": t.display_name,
# "status": t.status},
# tmp_list))}
res["result"] = True
except Exception as e:
raise e
return res
api_doc = {
"tags": ["登录日志"],
"parameters": [
{"name": "page_index",
"in": "query",
"type": "int",
"description": "当前页",
"default": 0},
{"name": "page_size",
"in": "query",
"type": "int",
"description": "条数",
"default": 1000},
{"name": "order_by",
"in": "query",
"type": "string",
"description": "排序"},
{"name": "filter_param",
"in": "query",
"type": "json",
"description": "过滤"}
],
"responses": {
200: {
"schema": {
"properties": {
}
}
}
}
}