Python

在使用 Pydantic 进行数据验证和设置时，你可能会遇到一个问题：当多个类有共同字段时，Pydantic 可能无法正确地区分这些类。让我通过一个简单的例子来解释这个问题。 ...

前因 由于python中没有方便快捷生成美观图片的方案，所以我使用了jinja2模板生成html后交给playwright生成网页图片，这其中，图片资源可以使用base64格式解决，而字体加载就难了。 ...

一般 try: import numpy except ImportError: print("numpy is not installed") 更推荐 from importlib.util import find_spec if find_spec("numpy") is not None: print("numpy is installed") else: print("numpy is not installed")

需求 以我的bot来说： Bot: ├─config ├─core ├─modules │ ├─required │ │ ├─BotManager │ │ ├─BotStatus │ │ ├─ExceptionCatch │ │ ├─... │ └─self │ ├─Modules1 │ ├─Modules2 │ ├─... ├─resources ├─service └─utils 模块统一放在modules下，required是bot运行必须的模块，self是可卸载安装的模块。 ...

描述 print(user) # name=user,age=18, ... session.commit() print(user) # empty 原因 sqlalchemy为了保持session内的数据同步，默认了expire_on_commit为True，它表现在： # 查询 b = session.query(User).filter(User.id==2).first() # 更改数据 b.name = "user_b" session.add(b) session.flush() session.commit() # commit后再次访问，其实是重新进行了select操作 b.name # user_b expire_on_commit可以用来更改 SQLAlchemy 的对象刷新机制，默认值为True即在session调用commit之后会主动将同一个session在commit之前查询得到的 ORM 对象的_sa_instance_state.expire属性设置为Flase，再次读取该对象属性时将重载这个对象，方法是重新调用之前的查询语句。 ...

连接单个数据库 from typing import Optional from sqlmodel import Field, Session, SQLModel, create_engine # 表 class Hero(SQLModel, table=True): id: Optional[int] = Field(default=None, primary_key=True) name: str secret_name: str age: Optional[int] = None # 连接 engine = create_engine("sqlite:///database.db") # 建表 SQLModel.metadata.create_all(engine) 连接多个数据库 from typing import Optional from sqlmodel import Field, Session, SQLModel, create_engine from sqlmodel.main import registry # 表1 table1 = registry() class Hero1(SQLModel, table=True, registry=table1): id: Optional[int] = Field(default=None, primary_key=True) name: str secret_name: str age: Optional[int] = None # 表2 table2 = registry() class Hero2(SQLModel, table=True, registry=table2): id: Optional[int] = Field(default=None, primary_key=True) name: str secret_name: str age: Optional[int] = None # 连接1 engine1 = create_engine("sqlite:///database1.db") # 连接2 engine2 = create_engine("sqlite:///database2.db") # 建表1 Hero1.metadata.create_all(engine1) # 建表2 Hero2.metadata.create_all(engine2) 一点优化 显然这样建多表会很麻烦，一种优化方案是： ...

同步 # sync_session # example(id, name) e = example(name="name1") db.add(e) db.flush() print(e.id) # id = 1 异步 # async_session # example(id, name) e = example(name="name1") db.add(e) await db.flush() await db.commit() await db.refresh(e) print(e.id) # id = 1

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 )