pip install japronto tortoise-orm
from tortoise import Tortoise, fields
from tortoise.models import Model
# 定义 Tortoise 配置
TORTOISE_ORM = {
'connections': {
'default': {
'engine': 'tortoise.backends.mysql',
'credentials': {
'host': '127.0.0.1',
'port': '3306',
'user': 'your_username',
'password': 'your_password',
'database': 'your_database',
}
}
},
'apps': {
'models': {
'models': ['__main__'],
'default_connection': 'default',
}
}
}
# 定义模型
class Item(Model):
id = fields.IntField(pk=True)
name = fields.CharField(unique=True, max_length=255)
# 初始化 Tortoise
async def initialize_db():
await Tortoise.init(TORTOISE_ORM)
await Tortoise.generate_schemas()
# 关闭 Tortoise
async def close_db():
await Tortoise.close_connections()
from japronto import Application
import json
app = Application()
# 初始化数据库
app.loop.create_task(initialize_db())
# 增删查改操作
async def get_item(request, item_id):
try:
item = await Item.get(id=item_id)
return json.dumps(item.__dict__)
except Item.DoesNotExist:
return request.Response(status_code=404)
async def create_item(request):
data = await request.json()
item = await Item.create(name=data['name'])
return json.dumps(item.__dict__), {'Content-Type': 'application/json'}
async def update_item(request, item_id):
data = await request.json()
try:
item = await Item.get(id=item_id)
item.name = data['name']
await item.save()
return json.dumps(item.__dict__)
except Item.DoesNotExist:
return request.Response(status_code=404)
async def delete_item(request, item_id):
try:
item = await Item.get(id=item_id)
await item.delete()
return request.Response(status_code=204)
except Item.DoesNotExist:
return request.Response(status_code=404)
# 定义路由
@app.route('/items/<int:item_id>')
async def handle_get(request, item_id):
return await get_item(request, item_id)
@app.route('/items', methods=['POST'])
async def handle_post(request):
return await create_item(request)
@app.route('/items/<int:item_id>', methods=['PUT'])
async def handle_put(request, item_id):
return await update_item(request, item_id)
@app.route('/items/<int:item_id>', methods=['DELETE'])
async def handle_delete(request, item_id):
return await delete_item(request, item_id)
# 程序入口
if __name__ == '__main__':
app.run(debug=True)
pip install PyJWT
import jwt
from japronto import Middleware
# 假设你的 JWT 密钥和验证逻辑如下
JWT_SECRET_KEY = 'your_secret_key'
ALGORITHM = 'HS256'
class AuthMiddleware(Middleware):
async def handle_request(self, request):
# 从请求头中获取 token
token = request.headers.get('Authorization', '').replace('Bearer ', '')
if token:
try:
# 解码 token 并获取用户信息
payload = jwt.decode(token, JWT_SECRET_KEY, algorithms=[ALGORITHM])
user_info = payload.get('user_info', {}) # 假设 JWT 中包含用户信息字段
# 将用户信息写入请求对象
request.user_info = user_info
except jwt.InvalidTokenError:
# 如果 token 无效，可以返回错误响应或继续处理请求
pass # 或者 request.respond(status_code=401)
# 继续请求的处理
await self.next(request)
from japronto import Application
app = Application()
# 将 AuthMiddleware 添加到应用中
app.add_middleware(AuthMiddleware)
# ... 其他路由和处理器 ...
# 示例处理器，使用请求中的 user_info
@app.route('/protected')
async def protected(request):
user_info = request.user_info
if user_info:
# 如果请求中存在用户信息，则处理请求
# 例如，返回用户信息
return json.dumps(user_info)
else:
# 如果没有用户信息，返回未授权
return request.Response(status_code=401)
# 程序入口
if __name__ == '__main__':
app.run(debug=True)
from japronto import Application
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.triggers.cron import CronTrigger
import asyncio
app = Application()
# 假设 AuthMiddleware 是已经定义好的中间件
app.add_middleware(AuthMiddleware)
# ... 其他路由和处理器 ...
@app.route('/protected')
async def protected(request):
user_info = request.user_info
if user_info:
return json.dumps(user_info)
else:
return request.Response(status_code=401)
# 定时任务
async def task_every_six_seconds():
print("每6秒处理一次的任务")
async def task_every_thirty_three_minutes():
print("每33分钟处理一次的任务")
async def task_daily_between_one_and_five():
print("每天早上1点到5点之间处理的任务")
# 初始化调度器
scheduler = AsyncIOScheduler()
# 添加定时任务
scheduler.add_job(task_every_six_seconds, IntervalTrigger(seconds=6))
scheduler.add_job(task_every_thirty_three_minutes, IntervalTrigger(minutes=33))
# 每天凌晨1点到5点之间每分钟执行一次
scheduler.add_job(task_daily_between_one_and_five, CronTrigger(hour=1, minute=0, end_hour=5, end_minute=59))
async def main():
# 启动调度器
scheduler.start()
# 启动 Japronto 应用
app.run(debug=True)
# 当 Japronto 应用停止时，等待定时任务完成
await scheduler.shutdown()
# 程序入口
if __name__ == '__main__':
asyncio.run(main())
pip install apscheduler
from japronto import Application
from japronto.middleware import Middleware
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.triggers.cron import CronTrigger
import asyncio
import json
# 定义跨域中间件
class CorsMiddleware(Middleware):
def __init__(self, app):
super().__init__(app)
async def handle(self, request):
# 设置跨域资源共享的响应头
response = await self.app(request)
response.headers['Access-Control-Allow-Origin'] = '*' # 允许来自所有源的请求
response.headers['Access-Control-Allow-Methods'] = 'GET, POST, PUT, DELETE, OPTIONS'
response.headers['Access-Control-Allow-Headers'] = 'Content-Type, Authorization'
response.headers['Access-Control-Allow-Credentials'] = 'true'
return response
app = Application()
# 添加跨域中间件
app.add_middleware(CorsMiddleware)
# ... 其他路由和处理器 ...
@app.route('/protected')
async def protected(request):
user_info = request.user_info
if user_info:
return json.dumps(user_info)
else:
return request.Response(status_code=401)
# 定时任务
async def task_every_six_seconds():
print("每6秒处理一次的任务")
async def task_every_thirty_three_minutes():
print("每33分钟处理一次的任务")
async def task_daily_between_one_and_five():
print("每天早上1点到5点之间处理的任务")
# 初始化调度器
scheduler = AsyncIOScheduler()
# 添加定时任务
scheduler.add_job(task_every_six_seconds, IntervalTrigger(seconds=6))
scheduler.add_job(task_every_thirty_three_minutes, IntervalTrigger(minutes=33))
# 每天凌晨1点到5点之间每分钟执行一次
scheduler.add_job(task_daily_between_one_and_five, CronTrigger(hour=1, minute=0, end_hour=5, end_minute=59))
async def main():
# 启动调度器
scheduler.start()
# 启动 Japronto 应用
app.run(debug=True)
# 当 Japronto 应用停止时，等待定时任务完成
await scheduler.shutdown()
# 程序入口
if __name__ == '__main__':
asyncio.run(main())
标签：__,经验,app,request,japronto,item,async,def From： https://www.cnblogs.com/pearlcity/p/17974979