knowledge/service/kg_prop_service.py

import traceback

from sqlalchemy.orm import Session
from typing import List
from model.kg_prop import KGProp
from db.session import get_db
import logging
from sqlalchemy.exc import IntegrityError

logger = logging.getLogger(__name__)

class KGPropService:
    def __init__(self, db: Session):
        self.db = db

    def get_props_by_ref_id(self, ref_id: int, prop_names: List[str] = None) -> List[dict]:
        try:
            query = self.db.query(KGProp).filter(KGProp.ref_id == ref_id)
            if prop_names:
                #prop_names是一个列表，需要使用in_方法进行查询
                query = query.filter(KGProp.prop_name.in_(prop_names))
            props = query.all()
            return [{
                'id': p.id,
                'category': p.category,
                'prop_name': p.prop_name,
                'prop_value': p.prop_value,
                'prop_title': p.prop_title,
                'type': p.type
            } for p in props]
        except Exception as e:
            logger.error(f"根据ref_id查询属性失败: {str(e)}")
            raise ValueError("查询失败")

    def get_prop_by_ref_id(self, ref_id: int, prop_name: str = None) -> List[dict]:
        try:
            query = self.db.query(KGProp).filter(KGProp.ref_id == ref_id)
            if prop_name:
                query = query.filter(KGProp.prop_name == prop_name)
            props = query.first()
            if not props:
                return None
            return {
                'id': props.id,
                'category': props.category,
                'prop_name': props.prop_name,
                'prop_value': props.prop_value,
                'prop_title': props.prop_title,
                'type': props.type
            }
        except Exception as e:
            traceback.print_exc()
            logger.error(f"根据id查询属性失败: {str(e)}")
            raise ValueError("查询失败")

    def create_prop(self, prop_data: dict) -> KGProp:
        try:
            new_prop = KGProp(**prop_data)
            self.db.add(new_prop)
            self.db.commit()
            return new_prop
        except IntegrityError as e:
            self.db.rollback()
            logger.error(f"创建属性失败: {str(e)}")
            raise ValueError("数据库完整性错误")

    def update_prop(self, prop_id: int, update_data: dict) -> KGProp:
        prop = self.db.query(KGProp).get(prop_id)
        if not prop:
            raise ValueError("属性未找到")

        try:
            for key, value in update_data.items():
                setattr(prop, key, value)
            self.db.commit()
            return prop
        except Exception as e:
            self.db.rollback()
            logger.error(f"更新属性失败: {str(e)}")
            raise ValueError("更新失败")

    def delete_prop(self, prop_id: int) -> None:
        prop = self.db.query(KGProp).get(prop_id)
        if not prop:
            raise ValueError("属性未找到")

        try:
            self.db.delete(prop)
            self.db.commit()
            return None
        except Exception as e:
            self.db.rollback()
            logger.error(f"删除属性失败: {str(e)}")
            raise ValueError("删除失败")