123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115 |
- 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"根据ref_id查询属性失败: {str(e)}")
- raise ValueError("查询失败")
- def get_prop_by_id(self, id: int)-> dict:
- try:
- query = self.db.query(KGProp).filter(KGProp.id == id)
- 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("删除失败")
|