kg_prop_service.py 3.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115
  1. import traceback
  2. from sqlalchemy.orm import Session
  3. from typing import List
  4. from model.kg_prop import KGProp
  5. from db.session import get_db
  6. import logging
  7. from sqlalchemy.exc import IntegrityError
  8. logger = logging.getLogger(__name__)
  9. class KGPropService:
  10. def __init__(self, db: Session):
  11. self.db = db
  12. def get_props_by_ref_id(self, ref_id: int, prop_names: List[str] = None) -> List[dict]:
  13. try:
  14. query = self.db.query(KGProp).filter(KGProp.ref_id == ref_id)
  15. if prop_names:
  16. #prop_names是一个列表,需要使用in_方法进行查询
  17. query = query.filter(KGProp.prop_name.in_(prop_names))
  18. props = query.all()
  19. return [{
  20. 'id': p.id,
  21. 'category': p.category,
  22. 'prop_name': p.prop_name,
  23. 'prop_value': p.prop_value,
  24. 'prop_title': p.prop_title,
  25. 'type': p.type
  26. } for p in props]
  27. except Exception as e:
  28. logger.error(f"根据ref_id查询属性失败: {str(e)}")
  29. raise ValueError("查询失败")
  30. def get_prop_by_ref_id(self, ref_id: int, prop_name: str = None) -> List[dict]:
  31. try:
  32. query = self.db.query(KGProp).filter(KGProp.ref_id == ref_id)
  33. if prop_name:
  34. query = query.filter(KGProp.prop_name == prop_name)
  35. props = query.first()
  36. if not props:
  37. return None
  38. return {
  39. 'id': props.id,
  40. 'category': props.category,
  41. 'prop_name': props.prop_name,
  42. 'prop_value': props.prop_value,
  43. 'prop_title': props.prop_title,
  44. 'type': props.type
  45. }
  46. except Exception as e:
  47. traceback.print_exc()
  48. logger.error(f"根据ref_id查询属性失败: {str(e)}")
  49. raise ValueError("查询失败")
  50. def get_prop_by_id(self, id: int)-> dict:
  51. try:
  52. query = self.db.query(KGProp).filter(KGProp.id == id)
  53. props = query.first()
  54. if not props:
  55. return None
  56. return {
  57. 'id': props.id,
  58. 'category': props.category,
  59. 'prop_name': props.prop_name,
  60. 'prop_value': props.prop_value,
  61. 'prop_title': props.prop_title,
  62. 'type': props.type
  63. }
  64. except Exception as e:
  65. traceback.print_exc()
  66. logger.error(f"根据id查询属性失败: {str(e)}")
  67. raise ValueError("查询失败")
  68. def create_prop(self, prop_data: dict) -> KGProp:
  69. try:
  70. new_prop = KGProp(**prop_data)
  71. self.db.add(new_prop)
  72. self.db.commit()
  73. return new_prop
  74. except IntegrityError as e:
  75. self.db.rollback()
  76. logger.error(f"创建属性失败: {str(e)}")
  77. raise ValueError("数据库完整性错误")
  78. def update_prop(self, prop_id: int, update_data: dict) -> KGProp:
  79. prop = self.db.query(KGProp).get(prop_id)
  80. if not prop:
  81. raise ValueError("属性未找到")
  82. try:
  83. for key, value in update_data.items():
  84. setattr(prop, key, value)
  85. self.db.commit()
  86. return prop
  87. except Exception as e:
  88. self.db.rollback()
  89. logger.error(f"更新属性失败: {str(e)}")
  90. raise ValueError("更新失败")
  91. def delete_prop(self, prop_id: int) -> None:
  92. prop = self.db.query(KGProp).get(prop_id)
  93. if not prop:
  94. raise ValueError("属性未找到")
  95. try:
  96. self.db.delete(prop)
  97. self.db.commit()
  98. return None
  99. except Exception as e:
  100. self.db.rollback()
  101. logger.error(f"删除属性失败: {str(e)}")
  102. raise ValueError("删除失败")