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