-
Notifications
You must be signed in to change notification settings - Fork 2
/
main.py
182 lines (143 loc) · 5.82 KB
/
main.py
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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
import os
import logging
import json
from sanic import Sanic
from sanic_cors import CORS
from contextvars import ContextVar
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import select
from apis.sanic_service_apis import sanic_service_apis
from model.engine import async_engine
from model.models import meta, User
from config import ProductionConfig
from utils.utils import error_return, protect_if_authenticated, is_revoked,get_language_data
from sanic_jwt import Initialize, exceptions
logging.basicConfig(
filename="/tmp/sanic_service.log",
filemode='a',
format=
'[%(asctime)s] [%(pathname)s] [%(lineno)d]: [%(levelname)s] [%(funcName)s]: %(message)s',
datefmt='%H:%M:%S',
level=logging.INFO)
def create_app(config=None):
"""
Creates and configures a Sanic application.
Parameters:
config (Optional[Dict[str, Any]]): A dictionary containing configuration options for the application. Defaults to None.
Returns:
Sanic: The configured Sanic application.
Middleware:
- inject_session: Middleware that injects a database session into the request context.
- close_session: Middleware that closes the database session after the response is sent.
Listeners:
- init_db: Listener that initializes the database and creates necessary tables if they don't exist.
"""
backend_app = Sanic("sanic_service", config=config)
backend_app.config.SECRET = os.urandom(32)
backend_app.blueprint(sanic_service_apis)
#
_base_model_session_ctx = ContextVar("session")
_info_sys_ctx = ContextVar("sanic_service")
CORS(backend_app)
@backend_app.middleware("request")
@protect_if_authenticated(config.JWT_WHITE_LIST)
async def inject_session(request):
"""
Middleware function that injects a session into the request's context.
Args:
request: The request object.
Returns:
None
"""
is_revoked(request, config.JWT_WHITE_LIST)
request.ctx.session = sessionmaker(bind=async_engine,
class_=AsyncSession,
expire_on_commit=False)()
request.ctx.session_ctx_token = _base_model_session_ctx.set(
request.ctx.session)
@backend_app.middleware("response")
async def close_session(request, response):
"""
Middleware function that is called after a response is sent.
:param request: The incoming request.
:param response: The outgoing response.
:return: None
"""
language = request.headers.get('Accept-Language') or 'zh'
# 中引文转义
if language == 'zh':
data = json.loads(response.body.decode('utf-8'))
data['msg'] = get_language_data(message=data['msg'],
language=language)
modified_response = json.dumps(data)
response.body = modified_response.encode('utf-8')
if hasattr(request.ctx, "session_ctx_token"):
_base_model_session_ctx.reset(request.ctx.session_ctx_token)
await request.ctx.session.close()
@backend_app.listener("after_server_start")
async def init_db(app):
"""
Initializes the database after the server starts.
Parameters:
- app: The application instance.
Returns:
None
"""
async with async_engine.begin() as conn:
await conn.run_sync(meta.create_all)
session = sessionmaker(bind=async_engine,
class_=AsyncSession,
expire_on_commit=False)()
async with session.begin():
user = (await
session.execute(select(User).where(User.name == "admin")
)).scalar()
if not user:
admin = User(name="admin", password="admin")
session.add(admin)
await session.close()
return backend_app
def generate_app_fun(config=None):
"""
Generate an application function.
:param config: Configuration options for creating the app (default: None)
:return: The generated app
"""
generate_app = create_app(config)
# init jwt authorized
Initialize(
generate_app,
authenticate=lambda username, password: password == 'password',
on_authenticate=lambda payload, request: setattr(
request.app.ctx, 'auth', payload),
on_payload_changed=lambda payload, request: setattr(
request.app.ctx, 'auth', payload),
secret=generate_app.config.JWT_SECRET,
expiration_delta=generate_app.config.JWT_ACCESS_TOKEN_EXPIRES,
user_id='id',
algorithm=generate_app.config.JWT_ALGORITHM,
)
# # intercept exceptions
@generate_app.exception(exceptions.SanicJWTException)
async def sanic_jwt_exception(request, exception):
"""
Handle SanicJWTException and return an error response.
Args:
request: The Sanic request object.
exception: The SanicJWTException object.
Returns:
An error response with the message and code from the exception.
Raises:
None.
"""
return error_return(message=exception.args[0],
code=exception.status_code)
return generate_app
if __name__ == "__main__":
app = generate_app_fun(ProductionConfig())
app.run(host="0.0.0.0",
port=8000,
access_log=False,
workers=int(os.environ.get("WORKER_NUM") or 4),
debug=False)