Jonathan
486749a890
Full-stack Dutch supermarket price tracker with FastAPI backend, PostgreSQL/SQLAlchemy, Albert Heijn scraper, and Next.js frontend. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
84 lines
3.9 KiB
Python
84 lines
3.9 KiB
Python
from datetime import datetime
|
|
|
|
import sqlalchemy as sa
|
|
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
|
|
|
from .database import Base
|
|
|
|
|
|
class Store(Base):
|
|
__tablename__ = "stores"
|
|
|
|
id: Mapped[int] = mapped_column(sa.Integer, primary_key=True)
|
|
name: Mapped[str] = mapped_column(sa.String(100), nullable=False)
|
|
slug: Mapped[str] = mapped_column(sa.String(50), unique=True, nullable=False)
|
|
country: Mapped[str] = mapped_column(sa.String(2), default="NL")
|
|
website: Mapped[str | None] = mapped_column(sa.String(255))
|
|
|
|
products: Mapped[list["Product"]] = relationship(back_populates="store")
|
|
scrape_runs: Mapped[list["ScrapeRun"]] = relationship(back_populates="store")
|
|
|
|
|
|
class Product(Base):
|
|
__tablename__ = "products"
|
|
__table_args__ = (
|
|
sa.UniqueConstraint("store_id", "external_id", name="uq_products_store_external"),
|
|
)
|
|
|
|
id: Mapped[int] = mapped_column(sa.Integer, primary_key=True)
|
|
store_id: Mapped[int] = mapped_column(sa.Integer, sa.ForeignKey("stores.id"), nullable=False)
|
|
external_id: Mapped[str] = mapped_column(sa.String(50), nullable=False)
|
|
ean: Mapped[str | None] = mapped_column(sa.String(20), index=True)
|
|
name: Mapped[str] = mapped_column(sa.String(255), nullable=False)
|
|
brand: Mapped[str | None] = mapped_column(sa.String(100))
|
|
category: Mapped[str | None] = mapped_column(sa.String(100))
|
|
unit_size: Mapped[str | None] = mapped_column(sa.String(50))
|
|
url: Mapped[str | None] = mapped_column(sa.String(500))
|
|
created_at: Mapped[datetime] = mapped_column(sa.DateTime, default=datetime.utcnow)
|
|
updated_at: Mapped[datetime] = mapped_column(
|
|
sa.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow
|
|
)
|
|
|
|
store: Mapped["Store"] = relationship(back_populates="products")
|
|
price_snapshots: Mapped[list["PriceSnapshot"]] = relationship(back_populates="product")
|
|
|
|
|
|
class ScrapeRun(Base):
|
|
__tablename__ = "scrape_runs"
|
|
|
|
id: Mapped[int] = mapped_column(sa.Integer, primary_key=True)
|
|
store_id: Mapped[int] = mapped_column(sa.Integer, sa.ForeignKey("stores.id"), nullable=False)
|
|
query: Mapped[str] = mapped_column(sa.String(255), nullable=False)
|
|
started_at: Mapped[datetime] = mapped_column(sa.DateTime, default=datetime.utcnow)
|
|
finished_at: Mapped[datetime | None] = mapped_column(sa.DateTime)
|
|
status: Mapped[str] = mapped_column(sa.String(20), default="running")
|
|
products_found: Mapped[int] = mapped_column(sa.Integer, default=0)
|
|
error_message: Mapped[str | None] = mapped_column(sa.String(1000))
|
|
|
|
store: Mapped["Store"] = relationship(back_populates="scrape_runs")
|
|
price_snapshots: Mapped[list["PriceSnapshot"]] = relationship(back_populates="scrape_run")
|
|
|
|
|
|
class PriceSnapshot(Base):
|
|
__tablename__ = "price_snapshots"
|
|
|
|
id: Mapped[int] = mapped_column(sa.Integer, primary_key=True)
|
|
product_id: Mapped[int] = mapped_column(
|
|
sa.Integer, sa.ForeignKey("products.id"), nullable=False
|
|
)
|
|
scrape_run_id: Mapped[int] = mapped_column(
|
|
sa.Integer, sa.ForeignKey("scrape_runs.id"), nullable=False
|
|
)
|
|
price: Mapped[int] = mapped_column(sa.Integer, nullable=False) # euro cents
|
|
unit_price: Mapped[int | None] = mapped_column(sa.Integer) # euro cents
|
|
unit_description: Mapped[str | None] = mapped_column(sa.String(50))
|
|
currency: Mapped[str] = mapped_column(sa.String(3), default="EUR")
|
|
discount_label: Mapped[str | None] = mapped_column(sa.String(100))
|
|
discount_description: Mapped[str | None] = mapped_column(sa.String(255))
|
|
was_price: Mapped[int | None] = mapped_column(sa.Integer) # euro cents, original price
|
|
is_on_sale: Mapped[bool] = mapped_column(sa.Boolean, default=False)
|
|
timestamp: Mapped[datetime] = mapped_column(sa.DateTime, default=datetime.utcnow, index=True)
|
|
|
|
product: Mapped["Product"] = relationship(back_populates="price_snapshots")
|
|
scrape_run: Mapped["ScrapeRun"] = relationship(back_populates="price_snapshots")
|