Add proper buy listing support with type-aware UI filters and display

crawler/repositories/listing_repository.py

@@ -1,5 +1,6 @@
 from datetime import datetime, timedelta
 import logging
+from typing import Generator
 from data_access import Listing
 from models.listing import (
     BuyListing,
@@ -9,13 +10,20 @@ from models.listing import (
     QueryParameters,
     RentListing,
 )
-from sqlalchemy import Engine
+from sqlalchemy import Engine, func, select as sa_select
 from sqlmodel import Session, select
 from sqlmodel.sql.expression import SelectOfScalar
 from tqdm import tqdm
 
 logger = logging.getLogger("uvicorn.error")
 
+# Columns needed for GeoJSON streaming (excludes routing_info_json, additional_info)
+STREAMING_COLUMNS = [
+    'id', 'price', 'number_of_bedrooms', 'square_meters',
+    'longitude', 'latitude', 'photo_thumbnail', 'last_seen',
+    'agency', 'price_history_json', 'available_from'
+]
+
 
 class ListingRepository:
     engine: Engine
@@ -58,6 +66,147 @@ class ListingRepository:
         logging.debug(f"Found {len(rows)} listings")
         return rows
 
+    def stream_listings(
+        self,
+        query_parameters: QueryParameters | None = None,
+        limit: int | None = None,
+        chunk_size: int = 100,
+    ) -> Generator[modelListing, None, None]:
+        """Yield listings one at a time for streaming.
+
+        Uses yield_per for memory-efficient iteration over large result sets.
+
+        Args:
+            query_parameters: Filtering parameters
+            limit: Maximum number of listings to yield
+            chunk_size: Number of rows to fetch at a time from the database
+        """
+        model = RentListing  # if no query params, default to renting listings
+        if query_parameters:
+            model = (
+                RentListing
+                if query_parameters.listing_type == ListingType.RENT
+                else BuyListing
+            )
+
+        query = select(model)
+        query = self._add_where_from_query_parameters(query, model, query_parameters)
+        if limit:
+            query = query.limit(limit)
+
+        with Session(self.engine) as session:
+            for listing in session.exec(query).yield_per(chunk_size):
+                yield listing
+
+    def _get_model_for_query(
+        self, query_parameters: QueryParameters | None
+    ) -> type[RentListing] | type[BuyListing]:
+        """Get the appropriate model class based on query parameters."""
+        if query_parameters and query_parameters.listing_type == ListingType.BUY:
+            return BuyListing
+        return RentListing
+
+    def count_listings(self, query_parameters: QueryParameters | None = None) -> int:
+        """Fast count for progress estimation."""
+        model = self._get_model_for_query(query_parameters)
+
+        query = sa_select(func.count(model.id))
+        query = self._add_where_from_query_parameters_raw(query, model, query_parameters)
+
+        with Session(self.engine) as session:
+            return session.execute(query).scalar() or 0
+
+    def stream_listings_optimized(
+        self,
+        query_parameters: QueryParameters | None = None,
+        limit: int | None = None,
+        page_size: int = 100,
+    ) -> Generator[dict, None, None]:
+        """Stream listings with keyset pagination and column projection.
+
+        Uses keyset pagination for O(1) performance at any offset, and only
+        fetches columns needed for GeoJSON (excludes large JSON blobs).
+
+        Args:
+            query_parameters: Filtering parameters
+            limit: Maximum number of listings to yield
+            page_size: Number of rows to fetch per database round-trip
+        """
+        model = self._get_model_for_query(query_parameters)
+
+        # Select only needed columns (excludes routing_info_json, additional_info)
+        columns = [
+            getattr(model, col) for col in STREAMING_COLUMNS if hasattr(model, col)
+        ]
+
+        last_id: int | None = None
+        total_yielded = 0
+
+        while True:
+            if limit and total_yielded >= limit:
+                break
+
+            query = sa_select(*columns)
+            query = self._add_where_from_query_parameters_raw(
+                query, model, query_parameters
+            )
+
+            # Keyset pagination: WHERE id > last_id (O(1) performance)
+            if last_id is not None:
+                query = query.where(model.id > last_id)
+
+            batch_limit = page_size
+            if limit:
+                batch_limit = min(page_size, limit - total_yielded)
+            query = query.order_by(model.id).limit(batch_limit)
+
+            with Session(self.engine) as session:
+                results = session.execute(query).fetchall()
+
+            if not results:
+                break
+
+            for row in results:
+                yield row._asdict()
+                last_id = row.id
+                total_yielded += 1
+
+            if len(results) < page_size:
+                break
+
+    def _add_where_from_query_parameters_raw(
+        self,
+        query,
+        model: type[RentListing] | type[BuyListing],
+        query_parameters: QueryParameters | None = None,
+    ):
+        """Add WHERE clauses from query parameters (for raw SQLAlchemy selects)."""
+        if query_parameters is None:
+            return query
+        query = query.where(
+            model.number_of_bedrooms.between(
+                query_parameters.min_bedrooms, query_parameters.max_bedrooms
+            ),
+            model.price.between(query_parameters.min_price, query_parameters.max_price),
+        )
+        if query_parameters.min_sqm is not None:
+            query = query.where(model.square_meters >= query_parameters.min_sqm)
+        if query_parameters.furnish_types and model == RentListing:
+            query = query.where(model.furnish_type.in_(query_parameters.furnish_types))
+        if (
+            model == RentListing
+            and query_parameters.let_date_available_from is not None
+        ):
+            query = query.where(
+                model.available_from >= query_parameters.let_date_available_from
+            )
+        if query_parameters.last_seen_days is not None:
+            last_seen_threshold = datetime.now() - timedelta(
+                days=query_parameters.last_seen_days
+            )
+            query = query.where(model.last_seen >= last_seen_threshold)
+        return query
+
     def _add_where_from_query_parameters(
         self,
         query: SelectOfScalar[Listing],
@@ -74,7 +223,7 @@ class ListingRepository:
         )
         if query_parameters.min_sqm is not None:
             query = query.where(model.square_meters >= query_parameters.min_sqm)
-        if query_parameters.furnish_types:
+        if query_parameters.furnish_types and model == RentListing:
             query = query.where(model.furnish_type.in_(query_parameters.furnish_types))
         if (
             isinstance(model, RentListing)