be67902b45
Request-level logging middleware (method/path/status/duration), per-module loggers in routes/services (nearby search radius expansion, 404s, name search), LOG_LEVEL setting wired through docker-compose, and seed_loader converted from print() to logging. Verified: pytest passes, and manually confirmed log lines appear for both successful and 404 requests via `docker compose logs api`. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
124 lines
4.2 KiB
Python
124 lines
4.2 KiB
Python
import logging
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from geoalchemy2 import Geometry
|
|
from sqlalchemy import and_, cast, func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.deps import lang_param, length_param
|
|
from app.db.session import get_db
|
|
from app.models.city import City
|
|
from app.models.enums import PlaceCategory
|
|
from app.models.place import Place
|
|
from app.models.place_content import PlaceContent
|
|
from app.schemas.common import LatLon
|
|
from app.schemas.place import ContentOut, PlaceDetail, PlaceListItem
|
|
from app.services.geocoding import search_places
|
|
|
|
logger = logging.getLogger("guidecity.places")
|
|
|
|
router = APIRouter(tags=["places"])
|
|
|
|
|
|
@router.get("/places/search", response_model=list[PlaceListItem])
|
|
def search(
|
|
q: str = Query(min_length=1),
|
|
city_slug: str | None = Query(default=None),
|
|
lang: str = Depends(lang_param),
|
|
db: Session = Depends(get_db),
|
|
) -> list[PlaceListItem]:
|
|
rows = search_places(db, q=q, language=lang, city_slug=city_slug)
|
|
return [
|
|
PlaceListItem(id=place.id, slug=place.slug, category=place.category, name=content.title, location=LatLon(lat=lat, lon=lon))
|
|
for place, content, lon, lat in rows
|
|
]
|
|
|
|
|
|
@router.get("/places/{place_id}", response_model=PlaceDetail)
|
|
def get_place(
|
|
place_id: int,
|
|
lang: str = Depends(lang_param),
|
|
length: str = Depends(length_param),
|
|
db: Session = Depends(get_db),
|
|
) -> PlaceDetail:
|
|
stmt = (
|
|
select(
|
|
Place,
|
|
PlaceContent,
|
|
func.ST_X(cast(Place.location, Geometry)).label("lon"),
|
|
func.ST_Y(cast(Place.location, Geometry)).label("lat"),
|
|
)
|
|
.join(
|
|
PlaceContent,
|
|
and_(
|
|
PlaceContent.place_id == Place.id,
|
|
PlaceContent.language == lang,
|
|
PlaceContent.length == length,
|
|
),
|
|
)
|
|
.where(Place.id == place_id)
|
|
)
|
|
row = db.execute(stmt).first()
|
|
if row is None:
|
|
logger.warning("place_id=%d not found for lang=%s length=%s", place_id, lang, length)
|
|
raise HTTPException(status_code=404, detail="Place not found")
|
|
|
|
place, content, lon, lat = row
|
|
return PlaceDetail(
|
|
id=place.id,
|
|
slug=place.slug,
|
|
category=place.category,
|
|
location=LatLon(lat=lat, lon=lon),
|
|
address=place.address,
|
|
district=place.district,
|
|
built_year=place.built_year,
|
|
architect_builder=place.architect_builder,
|
|
architectural_style=place.architectural_style,
|
|
content=ContentOut(language=content.language, length=content.length, title=content.title, body=content.body),
|
|
)
|
|
|
|
|
|
@router.get("/cities/{city_slug}/places", response_model=list[PlaceListItem])
|
|
def list_city_places(
|
|
city_slug: str,
|
|
category: PlaceCategory | None = Query(default=None),
|
|
district: str | None = Query(default=None),
|
|
q: str | None = Query(default=None),
|
|
lang: str = Depends(lang_param),
|
|
db: Session = Depends(get_db),
|
|
) -> list[PlaceListItem]:
|
|
city = db.scalar(select(City).where(City.slug == city_slug))
|
|
if city is None:
|
|
logger.warning("city_slug=%r not found", city_slug)
|
|
raise HTTPException(status_code=404, detail="City not found")
|
|
|
|
stmt = (
|
|
select(
|
|
Place,
|
|
PlaceContent,
|
|
func.ST_X(cast(Place.location, Geometry)).label("lon"),
|
|
func.ST_Y(cast(Place.location, Geometry)).label("lat"),
|
|
)
|
|
.join(
|
|
PlaceContent,
|
|
and_(
|
|
PlaceContent.place_id == Place.id,
|
|
PlaceContent.language == lang,
|
|
PlaceContent.length == "short",
|
|
),
|
|
)
|
|
.where(Place.city_id == city.id, Place.is_active.is_(True))
|
|
)
|
|
if category is not None:
|
|
stmt = stmt.where(Place.category == category)
|
|
if district is not None:
|
|
stmt = stmt.where(Place.district == district)
|
|
if q is not None:
|
|
stmt = stmt.where(PlaceContent.title.ilike(f"%{q}%"))
|
|
|
|
rows = db.execute(stmt).all()
|
|
logger.debug("city_slug=%r listing -> %d place(s)", city_slug, len(rows))
|
|
return [
|
|
PlaceListItem(id=p.id, slug=p.slug, category=p.category, name=pc.title, location=LatLon(lat=lat, lon=lon))
|
|
for p, pc, lon, lat in rows
|
|
]
|