mirror of
https://github.com/fastapi-practices/fastapi_best_architecture.git
synced 2025-08-18 23:11:48 +08:00
e43f128086
* add token refreshing mechanism * update token_expires to token_expire_time * Fix implicit type conversion exception catch
54 lines
1.3 KiB
Python
54 lines
1.3 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
import sys
|
|
|
|
from aioredis import Redis, TimeoutError, AuthenticationError
|
|
|
|
from backend.app.common.log import log
|
|
from backend.app.core.conf import settings
|
|
|
|
|
|
class RedisCli(Redis):
|
|
def __init__(self):
|
|
super(RedisCli, self).__init__(
|
|
host=settings.REDIS_HOST,
|
|
port=settings.REDIS_PORT,
|
|
password=settings.REDIS_PASSWORD,
|
|
db=settings.REDIS_DATABASE,
|
|
socket_timeout=settings.REDIS_TIMEOUT,
|
|
decode_responses=True, # 转码 utf-8
|
|
)
|
|
|
|
async def open(self):
|
|
"""
|
|
触发初始化连接
|
|
|
|
:return:
|
|
"""
|
|
try:
|
|
await self.ping()
|
|
except TimeoutError:
|
|
log.error('❌ 数据库 redis 连接超时')
|
|
sys.exit()
|
|
except AuthenticationError:
|
|
log.error('❌ 数据库 redis 连接认证失败')
|
|
sys.exit()
|
|
except Exception as e:
|
|
log.error('❌ 数据库 redis 连接异常 {}', e)
|
|
sys.exit()
|
|
|
|
async def delete_prefix(self, key: str):
|
|
"""
|
|
删除指定前缀的所有key
|
|
|
|
:param key:
|
|
:return:
|
|
"""
|
|
keys = await self.keys(f'{key}*')
|
|
if keys:
|
|
await self.delete(*keys)
|
|
|
|
|
|
# 创建redis连接对象
|
|
redis_client = RedisCli()
|