Python之数据库实体类作为参数动态传输

    def get_entity_class(self, entity_class_or_name):        """根据名称或类本身获取实体类"""        if isinstance(entity_class_or_name, str):            entity_class = self.entity_map.get(entity_class_or_name)            if not entity_class:                available_entities = ', '.join(self.entity_map.keys())                raise ValueError(                    f"Unknown entity class: {entity_class_or_name}. "                    f"Available entities: {available_entities}")            return entity_class        else:            return entity_class_or_name

    def query(self,              entity_class_or_name,              filters=None,              offset=0,              limit=None,              order_by=None,              distinct=False):        """动态查询方法"""        entity_class = self.get_entity_class(entity_class_or_name)        try:            """self.session在__init__(self)中进行初始化，self.session = get_default_db_service().get_db_session()"""            query = self.session.query(entity_class)            # 应用DISTINCT            if distinct:                query = query.distinct()            # 应用过滤条件            conditions = self.build_filters(entity_class, filters)            if conditions:                query = query.filter(and_(*conditions))            # 应用排序            if order_by:                if isinstance(order_by, str) and hasattr(                        entity_class, order_by):                    query = query.order_by(getattr(entity_class, order_by))                elif isinstance(order_by, (list, tuple)):                    order_columns = []                    for order_field in order_by:                        if isinstance(order_field, str) and hasattr(                                entity_class, order_field):                            order_columns.append(                                getattr(entity_class, order_field))                        elif hasattr(order_field, '__call__'):                            order_columns.append(order_field(entity_class))                    if order_columns:                        query = query.order_by(*order_columns)            # 应用分页            query = query.offset(offset)            if limit:                query = query.limit(limit)            return query.all()        finally:            self.session.close()

    def build_filters(self, entity_class, filters):        """构建SQLAlchemy过滤条件"""        if not filters:            return []        conditions = []        for key, value in filters.items():            if hasattr(entity_class, key):                if isinstance(value, dict):  # 复杂条件                    column = getattr(entity_class, key)                    if 'in' in value:                        conditions.append(column.in_(value['in']))                    elif 'like' in value:                        conditions.append(column.like(f"%{value['like']}%"))                    elif 'gt' in value:                        conditions.append(column > value['gt'])                    elif 'lt' in value:                        conditions.append(column < value['lt'])                    elif 'gte' in value:                        conditions.append(column >= value['gte'])                    elif 'lte' in value:                        conditions.append(column <= value['lte'])                else:  # 简单相等条件                    conditions.append(getattr(entity_class, key) == value)        return conditions

entity_service = DBEntityService()entity_service.register_entity("City", City)entity_list = entity_service.query("City",offset=0, limit=5)print('实体类查询所有结果：')for city in entity_list:    print(city.CITY_ID, city.CITY_NAME,city.REGION_ID)