selfrelease
8fe8047429
- 路由：GET/POST/PUT/DELETE /api/v1/companies - 支持分页、关键词搜索、行业/阶段筛选 - 租户隔离：只能操作本租户企业 - 测试：11 个企业 CRUD 测试，全部 passed（总计 26 tests）
137 lines
4.5 KiB
Python
137 lines
4.5 KiB
Python
"""企业档案路由：CRUD + 列表分页。"""
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.database import get_db
|
|
from app.core.dependencies import get_current_user
|
|
from app.models.company import Company
|
|
from app.models.user import User
|
|
from app.schemas.common import ApiResponse, success
|
|
from app.schemas.company import (
|
|
CompanyCreate,
|
|
CompanyListResponse,
|
|
CompanyResponse,
|
|
CompanyUpdate,
|
|
)
|
|
|
|
router = APIRouter(prefix="/companies", tags=["companies"])
|
|
|
|
|
|
@router.get("", response_model=ApiResponse[CompanyListResponse])
|
|
async def list_companies(
|
|
page: int = Query(default=1, ge=1),
|
|
page_size: int = Query(default=20, ge=1, le=100),
|
|
keyword: str | None = Query(default=None, description="按名称搜索"),
|
|
industry: str | None = Query(default=None, description="按行业筛选"),
|
|
stage: str | None = Query(default=None, description="按融资阶段筛选"),
|
|
db: AsyncSession = Depends(get_db),
|
|
user: User = Depends(get_current_user),
|
|
):
|
|
"""获取企业列表（分页 + 筛选）。"""
|
|
query = select(Company).where(Company.tenant_id == user.tenant_id)
|
|
|
|
if keyword:
|
|
query = query.where(Company.name.ilike(f"%{keyword}%"))
|
|
if industry:
|
|
query = query.where(Company.industry == industry)
|
|
if stage:
|
|
query = query.where(Company.stage == stage)
|
|
|
|
# 总数
|
|
count_query = select(func.count()).select_from(query.subquery())
|
|
total_result = await db.execute(count_query)
|
|
total = total_result.scalar_one()
|
|
|
|
# 分页
|
|
offset = (page - 1) * page_size
|
|
query = query.order_by(Company.created_at.desc()).offset(offset).limit(page_size)
|
|
result = await db.execute(query)
|
|
companies = result.scalars().all()
|
|
|
|
return success(
|
|
data=CompanyListResponse(
|
|
items=[CompanyResponse.model_validate(c, from_attributes=True) for c in companies],
|
|
total=total,
|
|
page=page,
|
|
page_size=page_size,
|
|
)
|
|
)
|
|
|
|
|
|
@router.get("/{company_id}", response_model=ApiResponse[CompanyResponse])
|
|
async def get_company(
|
|
company_id: str,
|
|
db: AsyncSession = Depends(get_db),
|
|
user: User = Depends(get_current_user),
|
|
):
|
|
"""获取企业详情。"""
|
|
result = await db.execute(
|
|
select(Company).where(Company.id == company_id, Company.tenant_id == user.tenant_id)
|
|
)
|
|
company = result.scalar_one_or_none()
|
|
if not company:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="企业不存在")
|
|
return success(data=CompanyResponse.model_validate(company, from_attributes=True))
|
|
|
|
|
|
@router.post("", response_model=ApiResponse[CompanyResponse], status_code=status.HTTP_201_CREATED)
|
|
async def create_company(
|
|
req: CompanyCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
user: User = Depends(get_current_user),
|
|
):
|
|
"""创建企业。"""
|
|
company = Company(tenant_id=user.tenant_id, **req.model_dump())
|
|
db.add(company)
|
|
await db.flush()
|
|
return success(
|
|
data=CompanyResponse.model_validate(company, from_attributes=True),
|
|
message="创建成功",
|
|
)
|
|
|
|
|
|
@router.put("/{company_id}", response_model=ApiResponse[CompanyResponse])
|
|
async def update_company(
|
|
company_id: str,
|
|
req: CompanyUpdate,
|
|
db: AsyncSession = Depends(get_db),
|
|
user: User = Depends(get_current_user),
|
|
):
|
|
"""更新企业信息。"""
|
|
result = await db.execute(
|
|
select(Company).where(Company.id == company_id, Company.tenant_id == user.tenant_id)
|
|
)
|
|
company = result.scalar_one_or_none()
|
|
if not company:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="企业不存在")
|
|
|
|
update_data = req.model_dump(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(company, key, value)
|
|
|
|
await db.flush()
|
|
return success(
|
|
data=CompanyResponse.model_validate(company, from_attributes=True),
|
|
message="更新成功",
|
|
)
|
|
|
|
|
|
@router.delete("/{company_id}", response_model=ApiResponse[None])
|
|
async def delete_company(
|
|
company_id: str,
|
|
db: AsyncSession = Depends(get_db),
|
|
user: User = Depends(get_current_user),
|
|
):
|
|
"""删除企业。"""
|
|
result = await db.execute(
|
|
select(Company).where(Company.id == company_id, Company.tenant_id == user.tenant_id)
|
|
)
|
|
company = result.scalar_one_or_none()
|
|
if not company:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="企业不存在")
|
|
|
|
await db.delete(company)
|
|
return success(message="删除成功")
|