feat: customer list API

Minimal read-only FastAPI surface over the mirrored DB:
- GET /api/customers with pagination (page/page_size) and optional
  channel / owner_account_id / department_id filters, ordered by
  last_chat_time desc (nulls last) then friend_id.
- GET /api/health liveness probe.
- Pydantic v2 CustomerOut + generic Page[T] envelope.
- CORS configured permissively for the local frontend.
- Tests override get_session to an in-memory async SQLite session
  (StaticPool) and drive the app via httpx ASGITransport.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
浅念
2026-08-01 12:10:03 +08:00
parent 5a2ff58af3
commit 1e0467b40f
7 changed files with 385 additions and 0 deletions
+6
View File
@@ -0,0 +1,6 @@
"""FastAPI layer for the yunke data-mirror viewing API.
Read-only HTTP surface over the mirrored DB. Public entrypoint is
:mod:`app.api.main` (``app = FastAPI(...)``). Routers live under
:mod:`app.api.routers` and share dependencies from :mod:`app.api.deps`.
"""
+10
View File
@@ -0,0 +1,10 @@
"""FastAPI dependencies for the yunke data-mirror API.
Re-exports :func:`app.db.base.get_session` so route code (and tests) depend on
a single API-layer seam. Tests override ``app.api.deps.get_session`` to swap in
an isolated in-memory SQLite session.
"""
from app.db.base import get_session
__all__ = ["get_session"]
+37
View File
@@ -0,0 +1,37 @@
"""FastAPI app for the yunke data-mirror viewing API.
Run locally with::
uvicorn app.api.main:app --reload
Includes:
* ``GET /api/health`` -> ``{"status": "ok"}`` liveness probe.
* ``GET /api/customers`` (see :mod:`app.api.routers.customer`).
CORS is configured permissively (wildcard origin, no credentials) so the local
frontend can call the API during development. Auth / finer CORS are LATER tasks.
"""
from __future__ import annotations
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.routers import customer
app = FastAPI(title="yunke data mirror")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(customer.router)
@app.get("/api/health")
async def health() -> dict[str, str]:
"""Trivial liveness probe."""
return {"status": "ok"}
+1
View File
@@ -0,0 +1 @@
"""HTTP routers for the yunke data-mirror API."""
+70
View File
@@ -0,0 +1,70 @@
"""GET /api/customers — paginated, filterable customer list.
Read-only query over the mirrored ``customer`` table. Supports filtering by
``channel``, ``owner_account_id`` (direct columns) and ``department_id``
(resolved through ``customer.user_id`` -> ``staff.dept_id``). Results are
ordered by ``last_chat_time desc`` (nulls last) then ``friend_id`` so the
most-recently-active contacts surface first.
"""
from __future__ import annotations
from typing import Annotated
import sqlalchemy as sa
from fastapi import APIRouter, Depends, Query
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.deps import get_session
from app.api.schemas import CustomerOut, Page
from app.db.models import Customer, Staff
router = APIRouter(prefix="/api", tags=["customers"])
SessionDep = Annotated[AsyncSession, Depends(get_session)]
PageParam = Annotated[int, Query(ge=1)]
PageSizeParam = Annotated[int, Query(ge=1, le=100)]
@router.get("/customers")
async def list_customers(
session: SessionDep,
page: PageParam = 1,
page_size: PageSizeParam = 20,
channel: Annotated[str | None, Query()] = None,
department_id: Annotated[str | None, Query()] = None,
owner_account_id: Annotated[str | None, Query()] = None,
) -> Page[CustomerOut]:
"""List customers with optional filters and pagination."""
conditions = []
if channel is not None:
conditions.append(Customer.channel == channel)
if owner_account_id is not None:
conditions.append(Customer.owner_account_id == owner_account_id)
if department_id is not None:
# Customer has no dept column; resolve via the owning staff member.
dept_user_ids = sa.select(Staff.user_id).where(
Staff.dept_id == department_id
)
conditions.append(Customer.user_id.in_(dept_user_ids))
total_stmt = sa.select(sa.func.count()).select_from(Customer)
if conditions:
total_stmt = total_stmt.where(*conditions)
total = (await session.execute(total_stmt)).scalar_one()
list_stmt = sa.select(Customer)
if conditions:
list_stmt = list_stmt.where(*conditions)
list_stmt = (
list_stmt.order_by(
sa.nullslast(Customer.last_chat_time.desc()),
Customer.friend_id.asc(),
)
.limit(page_size)
.offset((page - 1) * page_size)
)
rows = (await session.execute(list_stmt)).scalars().all()
items = [CustomerOut.model_validate(row) for row in rows]
return Page(items=items, total=total, page=page, page_size=page_size)
+43
View File
@@ -0,0 +1,43 @@
"""Pydantic v2 response models for the yunke data-mirror API."""
from __future__ import annotations
from typing import Generic, TypeVar
from pydantic import BaseModel, ConfigDict
T = TypeVar("T")
class CustomerOut(BaseModel):
"""Public Customer view — a subset of :class:`app.db.models.Customer`.
Only the fields the viewing frontend needs today are exposed; the verbatim
``raw_json`` blob and internal audit columns stay server-side.
"""
model_config = ConfigDict(from_attributes=True)
friend_id: str
channel: str
owner_account_id: str | None = None
friend_nickname: str | None = None
friend_remark: str | None = None
real_name: str | None = None
phone: str | None = None
region: str | None = None
sale_name: str | None = None
dept_name: str | None = None
head_image_url: str | None = None
last_chat_time: int | None = None
message_count: int | None = None
version: str | None = None
class Page(BaseModel, Generic[T]):
"""Generic pagination envelope returned by list endpoints."""
items: list[T]
total: int
page: int
page_size: int
+218
View File
@@ -0,0 +1,218 @@
"""Tests for app.api — minimal customer list API + health check.
Drives the FastAPI app over an in-process transport (``httpx.AsyncClient`` with
``ASGITransport``) — no real network socket. The ``get_session`` dependency is
overridden to yield an isolated in-memory async SQLite session. ``StaticPool``
keeps a single connection behind every session so all requests inside a test
see the same seeded schema/rows.
Coverage (Task 1.8 acceptance):
* ``GET /api/health`` -> 200 ``{"status": "ok"}``.
* ``GET /api/customers?channel=wechat`` -> ``total == 2`` and correct shape.
* ``GET /api/customers?channel=whatsapp`` -> ``total == 1``.
* Pagination: ``page_size=1&page=1`` -> 1 item, ``total`` still 3.
* Ordering: ``last_chat_time desc`` (nulls last) then ``friend_id``.
* Documented filters ``owner_account_id`` and ``department_id`` actually filter.
"""
from __future__ import annotations
import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import StaticPool
from app.api import deps
from app.api.main import app
from app.db.base import Base
from app.db.models import Customer, Staff
@pytest.fixture
async def client():
"""Yield an ``AsyncClient`` against ``app`` with a seeded in-memory DB.
Seeds three Customers (two wechat, one whatsapp) plus two Staff rows so the
``department_id`` filter (Customer.user_id -> Staff.dept_id) is exercisable.
"""
engine = create_async_engine(
"sqlite+aiosqlite:///:memory:",
poolclass=StaticPool,
connect_args={"check_same_thread": False},
)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
async with SessionLocal() as session:
session.add_all(
[
Customer(
friend_id="wx_a",
channel="wechat",
owner_account_id="wx_owner_1",
user_id="u1",
friend_nickname="Alice",
last_chat_time=2000,
message_count=10,
version="v1",
),
Customer(
friend_id="wx_b",
channel="wechat",
owner_account_id="wx_owner_1",
user_id="u2",
friend_nickname="Bob",
last_chat_time=1000,
version="v1",
),
Customer(
friend_id="wa_c",
channel="whatsapp",
owner_account_id="wa_owner_1",
user_id="u3",
friend_nickname="Carol",
last_chat_time=None,
version="v1",
),
Staff(user_id="u1", dept_id="d100"),
Staff(user_id="u2", dept_id="d200"),
]
)
await session.commit()
async def _override_session():
async with SessionLocal() as s:
yield s
app.dependency_overrides[deps.get_session] = _override_session
try:
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as ac:
yield ac
finally:
app.dependency_overrides.clear()
await engine.dispose()
async def test_health_returns_ok(client: AsyncClient):
"""``GET /api/health`` -> 200 ``{"status": "ok"}``."""
r = await client.get("/api/health")
assert r.status_code == 200
assert r.json() == {"status": "ok"}
async def test_list_default_page_returns_all(client: AsyncClient):
"""No query params -> page 1, page_size 20, all 3 rows."""
r = await client.get("/api/customers")
assert r.status_code == 200
body = r.json()
assert body["page"] == 1
assert body["page_size"] == 20
assert body["total"] == 3
assert len(body["items"]) == 3
# Shape: every documented CustomerOut field is present.
item = body["items"][0]
for key in (
"friend_id",
"channel",
"owner_account_id",
"friend_nickname",
"friend_remark",
"real_name",
"phone",
"region",
"sale_name",
"dept_name",
"head_image_url",
"last_chat_time",
"message_count",
"version",
):
assert key in item
async def test_list_filters_by_channel_wechat(client: AsyncClient):
"""``?channel=wechat`` -> total 2, both wechat rows."""
r = await client.get(
"/api/customers", params={"channel": "wechat", "page": 1, "page_size": 10}
)
assert r.status_code == 200
body = r.json()
assert body["total"] == 2
assert len(body["items"]) == 2
assert {it["friend_id"] for it in body["items"]} == {"wx_a", "wx_b"}
assert all(it["channel"] == "wechat" for it in body["items"])
async def test_list_filters_by_channel_whatsapp(client: AsyncClient):
"""``?channel=whatsapp`` -> total 1."""
r = await client.get("/api/customers", params={"channel": "whatsapp"})
assert r.status_code == 200
body = r.json()
assert body["total"] == 1
assert body["items"][0]["friend_id"] == "wa_c"
async def test_pagination_page_size_one(client: AsyncClient):
"""``page_size=1&page=1`` -> 1 item but ``total`` still reflects all rows."""
r = await client.get(
"/api/customers", params={"page": 1, "page_size": 1}
)
assert r.status_code == 200
body = r.json()
assert body["total"] == 3
assert len(body["items"]) == 1
async def test_order_is_last_chat_time_desc_nulls_last(client: AsyncClient):
"""Order: ``last_chat_time desc`` (nulls last), then ``friend_id``.
Seeded rows have last_chat_time 2000 / 1000 / None -> expected order
``[wx_a, wx_b, wa_c]`` (the whatsapp row with NULL last_chat_time lands
last even though it has no timestamp).
"""
r = await client.get(
"/api/customers", params={"page": 1, "page_size": 20}
)
assert r.status_code == 200
ids = [it["friend_id"] for it in r.json()["items"]]
assert ids == ["wx_a", "wx_b", "wa_c"]
async def test_filter_by_owner_account_id(client: AsyncClient):
"""``?owner_account_id=wa_owner_1`` -> only the whatsapp customer."""
r = await client.get(
"/api/customers", params={"owner_account_id": "wa_owner_1"}
)
assert r.status_code == 200
body = r.json()
assert body["total"] == 1
assert body["items"][0]["friend_id"] == "wa_c"
async def test_filter_by_department_id(client: AsyncClient):
"""``?department_id=d100`` -> customers whose ``user_id`` maps to that dept.
Customer ``wx_a`` has ``user_id=u1``; Staff ``u1`` belongs to ``d100``.
"""
r = await client.get(
"/api/customers", params={"department_id": "d100"}
)
assert r.status_code == 200
body = r.json()
assert body["total"] == 1
assert body["items"][0]["friend_id"] == "wx_a"
async def test_filters_compose_channel_and_owner(client: AsyncClient):
"""Two filters AND together: channel=wechat + owner=wx_owner_1 -> 2 rows."""
r = await client.get(
"/api/customers",
params={"channel": "wechat", "owner_account_id": "wx_owner_1"},
)
assert r.status_code == 200
body = r.json()
assert body["total"] == 2
assert {it["friend_id"] for it in body["items"]} == {"wx_a", "wx_b"}