疯狂的搬运工

Python 数据库插件 Sqlalchemy 为什么不能关闭连接？

为什么这段代码不会关闭数据库连接呢 "SELECT * FROM information_schema.PROCESSLIST;" from sqlalchemy import create_engine, URL, delete, update, select, exists from sqlalchemy.orm import sessionmaker, scoped_session from core.database.base import Base from lib.type import Type from typing import Any from flask import g, current_app import importlib import re class Database: ENV = None def set(self, key: str, value: any): """ @ 主要场景：令牌守卫设置[读取令牌下的] """ if self.ENV == "Application": g.Application = self.container._replace(**{key: value}) if self.ENV == 'Platform': g.Platform = self.container._replace(**{key: value}) @property def container(self): if self.ENV == "Application": if "Application" not in g: g.Application = Type.Application(None, None, None) return g.Application if self.ENV == 'Platform': if "Platform" not in g: g.Platform = Type.Platform(None, None) return g.Platform @property def database_conf(self): """ @ [平台-数据库]：配置 """ return Base.setting(current_app.config["Database"]) @property def __database_core(self): return self.__create_session(**self.database_conf) @property def __create_engine(self): """ @ 数据库引擎[缓存]：生成器 """ core = self.__database_core self.set("engine", core.engine) return core.engine @property def __create_database(self): """ @ 数据库[缓存]：生成器 """ core = self.__database_core self.set("database", core.session) return core.session def __create_session(self, **config): """ @ 创建容器：节约开销 """ engine = self.create_engine(**config) session = scoped_session(sessionmaker(bind=engine, autoflush=True)) return Type.Database(engine=engine, session=session()) @classmethod def create_engine(cls, **kwargs): """ @ 创建连接：引擎 """ return create_engine(URL.create("mysql+pymysql", **kwargs), echo=True, isolation_level="AUTOCOMMIT") @staticmethod def create_all(models: list, engine=None): """ @ 创建模型：批量 """ tables = [Database.get_model(model).__table__ for model in models] Base.metadata.create_all(bind=engine, tables=tables) def create_table(self, tables: list): Database.create_all(models=tables, engine=self.__create_engine) @staticmethod def get_model(model: str): """ @ 获取模型：对象实例 """ module = importlib.import_module(f"model.{model.split('_')[0]}.{model}") class_name = ''.join(re.findall(r"[a-zA-Z]+", model.split(".")[-1].title())) return getattr(module, class_name)() @property def database(self): """ @ 数据库[缓存] """ return self.__create_database if getattr(self.container, "database") is None else self.container.database def table_data_query_all(self, model: Any, condition: list = None, order: list = None, limit: int = 500, fields: list = None) -> list[dict]: """ @ 查询：多条 """ query = select(model) if fields is not None: query = query.with_only_columns(*fields) if condition is not None: query = query.filter(*condition) if order is not None: query = query.order_by(*order) asdasdas = [row.dict() for row in self.database.execute(query.limit(limit)).scalars()] self.database.get_bind().dispose() return asdasdas def table_data_query_one(self, model: Any, condition: list = None) -> dict: """ @ 查询：单条 """ result = self.database.execute(select(model).filter(*condition).limit(1)).scalar_one_or_none() return None if result is None else result.dict() def table_data_query_exists(self, condition: list) -> bool: """ @ 查询：存在 """ return self.database.query(exists().where(*condition)).scalar() def table_data_insert_all(self, models: list) -> None: """ @ 批量新增 """ with self.database as db: db.add_all(models) db.commit() def table_data_insert_one(self, model, data: bool = False) -> int | dict: """ @ 单个新增：默认返回自增主键：可选返回整条数据 """ with self.database as db: db.add(model) db.commit() return model.dict() if data is True else model.id def table_data_update(self, model: Any, condition: list, data: dict) -> None: """ @ 批量新增 """ with self.database as db: db.execute(update(model).where(*condition).values(**data)) def table_data_delete(self, model: Any, condition: list) -> None: """ @ 批量新增 """ with self.database as db: db.execute(delete(model).where(*condition)) def close(self): """ @ 关闭数据库连接 """ if self.database is not None: self.database.close() "image.png" (https://wmprod.oss-cn-shanghai.aliyuncs.com/c/user/20241013/5d8cbfbfb5b0b81d409ff9688ae9cdeb.png)

ZZZ葡萄皮

如何解决使用自定义装饰器时Pylance 类型检测错误？

我有一个使用自定义装饰器的 Python 代码片段，尽管代码成功运行，但在使用装饰的函数时，Pylance 一直显示警告。问题似乎与 Pylance 无法识别装饰的函数返回 Result[Any] 而不是 Select 有关。以下是代码： def execute(func): def inner_wrapper(*args, **kwargs) -> Result[Any]: with Session.begin() as session: result = session.execute(func(*args, **kwargs)) return result return inner_wrapper @execute def query_data_source( start_id: int = 1, max_results_amount: int = 10 ) -> Select: # 忽略 stmt = ( select( DataSource.id, DataSource.name, DataSource.source_url, DataSource.author, DataSource.description, DataSource.cover_image_url, DataSource.start_date, DataSource.end_date, ) .where(DataSource.id >= start_id) .limit(max_results_amount) .order_by(DataSource.id) ) return stmt 尽管代码功能正常，但 Pylance 发出警告，假设 query_data_source 函数返回 Select 而不是 Result[Any]。如何修改此代码以消除 Pylance 警告并准确传达装饰函数的返回类型？ 其中 Session 是 sqlalchemy.orm 的 session 对象。DataSouce 是一张数据库表。

ciiiiinema

Python 插件 sqlalchemy 循环创建表结果字段类型全部变成一样的奇怪问题？

from sqlalchemy import MetaData, Table, inspect, text, Column, Integer, String, DateTime, Date from flask import current_app from lib.error import Error from model.ide.ide_table import IdeTable from model.ide.ide_attr import IdeAttr from model.ide.ide_section import IdeSection from core.environment.application import Application class BuilderTable(Application): def create(self, module: str, table: str, name: str, parent: int = None, back_data: bool = True) -> int | dict: """ @ SqlAlchemy：生成：数据表 :param module: 模块名称 :param table: 表名称 :param name: 表显示名称 :param parent: 父表 :param back_data: 是否返回数据 """ if inspect(self.engine).has_table(table): raise Error(f"{table}：{Error.TABLE_ALREADY_EXISTS}") fields_default = "这里是 yaml 配置" if parent is None: fields_default = [field for field in fields_default if field["name"] != "fid"] fields_format = [self.generate_column(field) for field in fields_default] Table(table, MetaData(), *fields_format).create(self.engine) return _table @classmethod def generate_column(cls, params: dict): """ @ SqlAlchemy：生成：字段对象 :param params: 字段参数 """ format_data = params if params.get("type_") is not None: format_data["type_"] = cls.generate_type(params["type_"]) if params.get("server_default") is not None: format_data["server_default"] = text(str(params["server_default"])) return Column(**format_data) @classmethod def generate_type(cls, typer: str, length: int = 50): """ @ SqlAlchemy：生成：类型对象 :param typer: 类型 :param length: 长度 """ if typer == "int": return Integer elif typer == "str": return String(length) elif typer == "date": return Date elif typer == "datetime": return DateTime else: return String(length) 配置是 "yaml" field: # 默认字段：新建模块场景 default: - name: id type_: int comment: 系统编号 primary_key: true - name: sid type_: varchar comment: 业务编号 index: true primary_key: false - name: fid type_: int comment: 父表编号 index: true primary_key: false - name: create_user type_: int comment: 创建人 index: true primary_key: false - name: create_group type_: int comment: 创建组 index: true primary_key: false - name: update_user type_: int comment: 修改人 index: true primary_key: false - name: update_group type_: int comment: 修改组 index: true primary_key: false - name: create_time type_: datetime comment: 创建时间 primary_key: false - name: update_time type_: datetime comment: 更新时间 primary_key: false - name: status type_: int comment: 系统状态 index: true primary_key: false server_default: 1 然后代外层循环创建表的时候居然除了第一张表是对的后面全是 "varchar" 了，人都傻了，这怎么回事 "image.png" (https://wmprod.oss-cn-shanghai.aliyuncs.com/c/user/20240925/28c03e680821b78080815ec6940a4ed4.png)

练习代码二十年

Python SQLAlchemy 怎么对表的字段进行：添加、修改、删除？

"SQLAlchemy" 怎么对表的字段进行：添加、修改、删除？除了直接写原生 "SQL" 以外，现在创建表很容易单我没找到修改表的 "alter" 方法？有谁用过么？