adai/MemoryBear
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[changes] user_alias修改为end_user_info

Browse Source
This commit is contained in:
lanceyq
2026-03-24 17:30:47 +08:00
parent e8d575fd0b
commit db14d40fb3
8 changed files with 257 additions and 257 deletions
Download Patch File Download Diff File Expand all files Collapse all files

90
api/app/repositories/end_user_info_repository.py Normal file
View File

@@ -0,0 +1,90 @@
"""
终端用户信息仓储层
"""
import uuid
from typing import List, Optional
from sqlalchemy.orm import Session
from app.models.end_user_info_model import EndUserInfo
from app.core.logging_config import get_logger
logger = get_logger(__name__)
class EndUserInfoRepository:
"""终端用户信息仓储类"""
def __init__(self, db: Session):
self.db = db
def create(self, end_user_id: uuid.UUID, other_name: str, alias: str = None, meta_data: dict = None) -> EndUserInfo:
"""创建终端用户信息"""
end_user_info = EndUserInfo(
end_user_id=end_user_id,
other_name=other_name,
alias=alias,
meta_data=meta_data
)
self.db.add(end_user_info)
self.db.commit()
self.db.refresh(end_user_info)
logger.info(f"创建终端用户信息: end_user_id={end_user_id}, alias={alias}")
return end_user_info
def get_by_id(self, info_id: uuid.UUID) -> Optional[EndUserInfo]:
"""根据ID获取用户信息"""
return self.db.query(EndUserInfo).filter(EndUserInfo.id == info_id).first()
def get_by_end_user_id(self, end_user_id: uuid.UUID) -> List[EndUserInfo]:
"""获取用户的所有信息记录"""
return self.db.query(EndUserInfo).filter(EndUserInfo.end_user_id == end_user_id).all()
def update(self, info_id: uuid.UUID, alias: str = None, meta_data: dict = None) -> Optional[EndUserInfo]:
"""更新用户信息"""
end_user_info = self.get_by_id(info_id)
if end_user_info:
if alias is not None:
end_user_info.alias = alias
if meta_data is not None:
end_user_info.meta_data = meta_data
self.db.commit()
self.db.refresh(end_user_info)
logger.info(f"更新终端用户信息: info_id={info_id}")
return end_user_info
def delete(self, info_id: uuid.UUID) -> bool:
"""删除用户信息"""
end_user_info = self.get_by_id(info_id)
if end_user_info:
self.db.delete(end_user_info)
self.db.commit()
logger.info(f"删除终端用户信息: info_id={info_id}")
return True
return False
def delete_by_end_user_id(self, end_user_id: uuid.UUID) -> int:
"""删除用户的所有信息记录"""
count = self.db.query(EndUserInfo).filter(EndUserInfo.end_user_id == end_user_id).delete()
self.db.commit()
logger.info(f"删除用户所有信息记录: end_user_id={end_user_id}, count={count}")
return count
def batch_create(self, end_user_id: uuid.UUID, other_name: str, aliases: List[str]) -> List[EndUserInfo]:
"""批量创建用户信息"""
end_user_infos = []
for alias in aliases:
if alias and alias.strip():
end_user_info = EndUserInfo(
end_user_id=end_user_id,
other_name=other_name,
alias=alias.strip()
)
self.db.add(end_user_info)
end_user_infos.append(end_user_info)
self.db.commit()
for end_user_info in end_user_infos:
self.db.refresh(end_user_info)
logger.info(f"批量创建终端用户信息: end_user_id={end_user_id}, count={len(end_user_infos)}")
return end_user_infos

90
api/app/repositories/user_alias_repository.py
View File

@@ -1,90 +0,0 @@
"""
用户别名仓储层
"""
import uuid
from typing import List, Optional
from sqlalchemy.orm import Session
from app.models.user_alias_model import UserAlias
from app.core.logging_config import get_logger
logger = get_logger(__name__)
class UserAliasRepository:
"""用户别名仓储类"""
def __init__(self, db: Session):
self.db = db
def create(self, end_user_id: uuid.UUID, other_name: str, alias: str = None, meta_data: dict = None) -> UserAlias:
"""创建用户别名"""
user_alias = UserAlias(
end_user_id=end_user_id,
other_name=other_name,
alias=alias,
meta_data=meta_data
)
self.db.add(user_alias)
self.db.commit()
self.db.refresh(user_alias)
logger.info(f"创建用户别名: end_user_id={end_user_id}, alias={alias}")
return user_alias
def get_by_id(self, alias_id: uuid.UUID) -> Optional[UserAlias]:
"""根据ID获取别名"""
return self.db.query(UserAlias).filter(UserAlias.id == alias_id).first()
def get_by_end_user_id(self, end_user_id: uuid.UUID) -> List[UserAlias]:
"""获取用户的所有别名"""
return self.db.query(UserAlias).filter(UserAlias.end_user_id == end_user_id).all()
def update(self, alias_id: uuid.UUID, alias: str = None, meta_data: dict = None) -> Optional[UserAlias]:
"""更新别名"""
user_alias = self.get_by_id(alias_id)
if user_alias:
if alias is not None:
user_alias.alias = alias
if meta_data is not None:
user_alias.meta_data = meta_data
self.db.commit()
self.db.refresh(user_alias)
logger.info(f"更新用户别名: alias_id={alias_id}")
return user_alias
def delete(self, alias_id: uuid.UUID) -> bool:
"""删除别名"""
user_alias = self.get_by_id(alias_id)
if user_alias:
self.db.delete(user_alias)
self.db.commit()
logger.info(f"删除用户别名: alias_id={alias_id}")
return True
return False
def delete_by_end_user_id(self, end_user_id: uuid.UUID) -> int:
"""删除用户的所有别名"""
count = self.db.query(UserAlias).filter(UserAlias.end_user_id == end_user_id).delete()
self.db.commit()
logger.info(f"删除用户所有别名: end_user_id={end_user_id}, count={count}")
return count
def batch_create(self, end_user_id: uuid.UUID, other_name: str, aliases: List[str]) -> List[UserAlias]:
"""批量创建别名"""
user_aliases = []
for alias in aliases:
if alias and alias.strip():
user_alias = UserAlias(
end_user_id=end_user_id,
other_name=other_name,
alias=alias.strip()
)
self.db.add(user_alias)
user_aliases.append(user_alias)
self.db.commit()
for user_alias in user_aliases:
self.db.refresh(user_alias)
logger.info(f"批量创建用户别名: end_user_id={end_user_id}, count={len(user_aliases)}")
return user_aliases