Add POI CLI commands: add-poi, list-pois, calculate-poi

Enables POI management and distance calculation from the command line,
following the existing Click command patterns with asyncio.run().
This commit is contained in:
Viktor Barzin 2026-02-08 13:15:24 +00:00
parent bd788df9aa
commit 149311508e
No known key found for this signature in database
GPG key ID: 0EB088298288D958

90
main.py
View file

@ -16,7 +16,10 @@ from services import (
listing_service, listing_service,
export_service, export_service,
district_service, district_service,
poi_service,
) )
from repositories.poi_repository import POIRepository
from repositories.user_repository import UserRepository
P = ParamSpec("P") P = ParamSpec("P")
R = TypeVar("R") R = TypeVar("R")
@ -380,5 +383,92 @@ def list_districts() -> None:
click.echo(f" - {name}") click.echo(f" - {name}")
@cli.command()
@click.option("--name", required=True, help="Name for the POI (e.g., 'Office')")
@click.option("--address", required=True, help="Human-readable address")
@click.option("--lat", required=True, type=float, help="Latitude")
@click.option("--lon", required=True, type=float, help="Longitude")
@click.option("--user-email", required=True, help="User email to associate POI with")
def add_poi(name: str, address: str, lat: float, lon: float, user_email: str) -> None:
"""Create a Point of Interest."""
user_repo = UserRepository(engine)
db_user = user_repo.get_user_by_email(user_email)
if db_user is None:
db_user = user_repo.create_user(user_email)
poi_repo = POIRepository(engine)
result = poi_service.create_poi(
poi_repo,
user_id=db_user.id, # type: ignore[arg-type]
name=name,
address=address,
latitude=lat,
longitude=lon,
)
click.echo(f"Created POI: {result.poi.name} (id={result.poi.id})")
@cli.command()
@click.option("--user-email", required=True, help="User email to list POIs for")
def list_pois(user_email: str) -> None:
"""List POIs for a user."""
user_repo = UserRepository(engine)
db_user = user_repo.get_user_by_email(user_email)
if db_user is None:
click.echo(f"User '{user_email}' not found")
return
poi_repo = POIRepository(engine)
pois = poi_service.get_user_pois(poi_repo, db_user.id) # type: ignore[arg-type]
if not pois:
click.echo("No POIs found")
return
for p in pois:
click.echo(f" [{p.id}] {p.name} - {p.address} ({p.latitude}, {p.longitude})")
@cli.command()
@click.option("--poi-id", required=True, type=int, help="POI ID to calculate distances for")
@click.option(
"--travel-modes",
required=True,
help="Comma-separated travel modes (WALK, BICYCLE, TRANSIT)",
)
@click.option(
"--listing-type",
required=True,
type=click.Choice(ListingType.__members__.keys(), case_sensitive=False),
help="Listing type (BUY or RENT)",
)
def calculate_poi(poi_id: int, travel_modes: str, listing_type: str) -> None:
"""Calculate distances from listings to a POI."""
from services.poi_distance_calculator import calculate_poi_distances as calc
from config.routing_config import RoutingConfig
poi_repo = POIRepository(engine)
poi = poi_repo.get_poi_by_id(poi_id)
if poi is None:
click.echo(f"POI {poi_id} not found")
return
listing_repo = ListingRepository(engine=engine)
lt = ListingType[listing_type]
modes = [m.strip().upper() for m in travel_modes.split(",")]
click.echo(f"Calculating {modes} distances for POI '{poi.name}' ({lt.value} listings)...")
total = asyncio.run(calc(
listing_repo=listing_repo,
poi_repo=poi_repo,
poi=poi,
travel_modes=modes,
listing_type=lt,
config=RoutingConfig.from_env(),
))
click.echo(f"Computed {total} distances")
if __name__ == "__main__": if __name__ == "__main__":
cli() cli()