Spaces:
Sleeping
Sleeping
File size: 4,060 Bytes
551658a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 | from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from database import get_db
from models.contact import Contact
from schemas.contact import ContactCreate, ContactUpdate, ContactResponse
from routers.auth import get_current_user, is_admin_user
from models.user import User
router = APIRouter()
# 获取公司的联系信息列表
@router.get("/company/{company_code}", response_model=List[ContactResponse])
async def get_company_contacts(
company_code: str,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
):
# 检查用户权限：admin用户和超级管理员可以访问所有公司，普通用户只能访问自己公司
if not is_admin_user(current_user) and not current_user.is_superadmin and current_user.company_code != company_code:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have permission to access this company's contacts"
)
contacts = db.query(Contact).filter(Contact.company_code == company_code).all()
return contacts
# 创建联系信息
@router.post("/company/{company_code}", response_model=ContactResponse)
async def create_contact(
contact: ContactCreate,
company_code: str,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
):
# 检查用户权限：admin用户和超级管理员可以访问所有公司，普通用户只能访问自己公司
if not is_admin_user(current_user) and not current_user.is_superadmin and current_user.company_code != company_code:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have permission to create contacts for this company"
)
db_contact = Contact(
**contact.model_dump(),
company_code=company_code
)
db.add(db_contact)
db.commit()
db.refresh(db_contact)
return db_contact
# 更新联系信息
@router.put("/{contact_id}", response_model=ContactResponse)
async def update_contact(
contact_id: int,
contact: ContactUpdate,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
):
db_contact = db.query(Contact).filter(Contact.id == contact_id).first()
if not db_contact:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Contact not found"
)
# 检查用户权限：admin用户和超级管理员可以访问所有公司，普通用户只能访问自己公司
if not is_admin_user(current_user) and not current_user.is_superadmin and db_contact.company_code != current_user.company_code:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have permission to update this contact"
)
update_data = contact.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(db_contact, field, value)
db.commit()
db.refresh(db_contact)
return db_contact
# 删除联系信息
@router.delete("/{contact_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_contact(
contact_id: int,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
):
db_contact = db.query(Contact).filter(Contact.id == contact_id).first()
if not db_contact:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Contact not found"
)
# 检查用户权限：admin用户和超级管理员可以访问所有公司，普通用户只能访问自己公司
if not is_admin_user(current_user) and not current_user.is_superadmin and db_contact.company_code != current_user.company_code:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have permission to delete this contact"
)
db.delete(db_contact)
db.commit()
return None
|