归档

PDM介绍 一个现代的 Python 包管理器，支持 PEP 582 跟pipenv与poetry有着相同功能 抛弃传统包管理器所需的虚拟环境 - PEP 582 比poetry更快的依赖解析 类似pnpm的中心化安装缓存 插件系统 PEP 621 元数据格式 兼容 PEP 517 的构建后端，用于构建发布包(源码格式与 wheel 格式) ...

基本的上下文管理器 需要定义： class ContextM: def __enter__(): dosomethings() def __exit__(): dosometings() class AsyncContextM: async def __aenter__(): await dosometings() async def __aexit__(): await dosometings() 调用时使用with ContextM()或async with AsyncContextM ContextManager Python内置的一个很方便的生成上下文管理器的装饰器 ...

日志重写 import logging import os import sys from pprint import pformat from loguru import logger from config import settings class InterceptHandler(logging.Handler): def emit(self, record): # Get corresponding Loguru level if it exists try: level = logger.level(record.levelname).name except ValueError: level = record.levelno # Find caller from where originated the logged message frame, depth = logging.currentframe(), 2 while frame.f_code.co_filename == logging.__file__: frame = frame.f_back depth += 1 logger.opt(depth=depth, exception=record.exc_info).log( level, record.getMessage() ) def format_record(record: dict) -> str: format_string = settings.config.LOG_FORMAT if record["extra"].get("payload") is not None: record["extra"]["payload"] = pformat( record["extra"]["payload"], indent=4, compact=True, width=88 ) format_string += "\n<level>{extra[payload]}</level>" format_string += "{exception}\n" return format_string # it requires app init or won't be work def logger_init(): log_path_all = os.path.join(settings.config.log_path, settings.config.LOG_FILE) log_path_error = os.path.join(settings.config.log_error_path, settings.config.LOG_FILE) logging.getLogger().handlers = [InterceptHandler()] logger.configure( handlers=[{"sink": sys.stdout, "level": logging.DEBUG, "format": format_record}]) # logger.add(config.LOG_DIR, encoding='utf-8', rotation="9:46") logger.add(log_path_all, rotation="00:00", retention="2 week", enqueue=True, diagnose=True, level="DEBUG", format=format_record ) logger.add(log_path_error, rotation="00:00", retention="3 week", enqueue=True, backtrace=True, diagnose=True, level="ERROR", format=format_record ) logger.debug('日志系统已加载\n\n\n') logging.getLogger("uvicorn.access").handlers = [InterceptHandler()] 在app中调用 app = FastAPI( title=config.settings.title, debug=config.settings.config.DEBUG ) # app create frist, then init logger logger_init() origins = [ "http://127.0.0.1", f"http://127.0.0.1:{config.settings.config.port}", "http://localhost", f"http://localhost:{config.settings.config.port}", # "*" ] app.include_router(scraper.router) app.include_router(works.router) app.include_router(users.router) app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) if __name__ == '__main__': uvicorn.run( app='main:app', host='0.0.0.0', port=config.settings.config.port, reload=True, # should be False at production access_log=True, ws_ping_interval=300, ws_ping_timeout=300 )

以Depends的方式定义jwt from jose import JWTError, jwt from passlib.context import CryptContext from fastapi.security import OAuth2PasswordBearer from fastapi import Depends, HTTPException, status from db import Curd from typing import Optional from datetime import datetime from config import settings from utils import logger from .db import get_db from datetime import timedelta pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") oauth2_schema = OAuth2PasswordBearer(tokenUrl="/users/login") def hash_password(plain_password: str) -> str: return pwd_context.hash(plain_password) def verity_password(plain_password: str, hashed_password: str) -> bool: return pwd_context.verify(plain_password, hashed_password) async def jwt_authenticate_user(username: str, password: str): db = get_db() user = await Curd.get_user_by_name(await db.__anext__(), username) if not user: return False if not verity_password(password, user.password): return False return user def created_access_token(data: dict, expires_delta: Optional[timedelta] = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=settings.config.ACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(claims=to_encode, key=settings.default.SECRET_KEY, algorithm="HS256") return encoded_jwt async def jwt_verity(token: str = Depends(oauth2_schema)): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: playload = jwt.decode(token=token, key=settings.default.SECRET_KEY, algorithms=["HS256"]) except JWTError as e: logger.error("验证出现错误:\n", e) raise credentials_exception else: return playload async def jwt_get_current_user(playload: dict = Depends(jwt_verity)): db = get_db() user_id = int(playload.get("sub")) user = await Curd.get_user_by_id(await db.__anext__(), user_id=user_id) return user async def jwt_verity_admin(playload: dict = Depends(jwt_verity)): db = get_db() user_id = int(playload.get("sub")) is_admin = playload.get("is_admin") if is_admin: user = await Curd.get_user_by_id(await db.__anext__(), user_id=user_id) if user.is_admin: return raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) 需要分发token的url router = APIRouter( prefix="/users", tags=["user"], ) # it defind at dependencies.jwt # oauth2_schema = OAuth2PasswordBearer(tokenUrl="/users/token") @router.post("/register", response_model=UserToken) async def user_register(user: OAuth2PasswordRequestForm = Depends(), db: AsyncSession = Depends(get_db)): if await Curd.is_user_name_exists(db, user.username): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="username already exists") hashed_password = hash_password(user.password) new_user = UserCreate(id=worker.get_id(), name=user.username, password=hashed_password) await Curd.register_user(db, new_user) @router.post("/login", response_model=UserToken) async def user_login(user: OAuth2PasswordRequestForm = Depends()): user = await jwt_authenticate_user(username=user.username, password=user.password) if not user: raise HTTPException( status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password", headers={"WWW-Authenticate": "Bearer"}, ) access_token = created_access_token(data={"sub": str(user.id)}) user_token = UserToken(username=user.name, access_token=access_token, token_type="bearer") return user_token 需要token进行访问限制的url router = APIRouter( prefix="/works", tags=["works"], ) @router.get("/get", response_model=Works) async def test( order: int, nsfw: bool, page: int, sort: int, user: User = Depends(jwt_get_current_user), db: AsyncSession = Depends(get_db), ): works = await Curd.get_works(db, user.id, order, nsfw, page, sort) return works

ORM ORM：对象关系映射（Object Relational Mapping，简称ORM，或O/RM，或O/R mapping)，用于实现面向对象编程语言里不同类型系统的数据之间的转换 ...

1.什么是算法？ 算法是计算机科学中的基本概念，是对特定问题的求解步骤 2.什么是算法的复杂度？ 分为时间复杂度和空间复杂度，前者一般是指程序运行从开始到结束所需的时间，后者一般是指对应的程序从运行开始到结束所需的储存空间 ...

前言 Pycharm好像是3.X版本上的登录验证，IEDA应该是2.3，登录验证已经无法使用插件来绕过，这里记录一下新的方法。 此方法来源依旧是eval-reset插件的作者，使用到的开源项目地址或网站： ...

Linux FTP Docker Apache 基本的排错 检查端口是否打开 检查服务/容器是否启动 检查防火墙是否关闭 ftp中检查配置文件中的 ip 是否替换为了自己的ip docker中注意容器的命名冲突

前言 Apache程序是目前拥有很高市场占有率的Web服务程序之一，这里演示如何安装，并访问一个简单的index.html 开整 首先安装apache的软件包： ...

安装 # 安装 yum install vsftpd -y yum install ftp -y # 启用 systemctl restart vsftpd systemctl enable vsftpd # 处理配置文件 mv /etc/vsftpd/vsftpd.conf /etc/vsftpd/vsftpd.conf.bak grep -v "#" /etc/vsftpd/vsftpd.conf.bak > /etc/vsftpd/vsftpd.conf # 关闭SELinux setenforce 0 修改配置文件： vim /etc/vsftpd/vsftpd.conf 将以下内容复制进去（注意！将18行 pasv_address 等号后面的ip改为你自己云服务器的ip地址！！） ...