grokbot_ty
e5514fe3bf
- App factory pattern (FUND-8): create_app() produces fresh app per test
- Logging config (structured, no secrets)
- Config validation: SLAB_AES_KEY required, WOO_BASE_URL HTTPS-only
- GET /api/health returns 200 {status: ok, version, woo_reachable: false}
- SQLite + alembic skeleton: full DATA-MODEL schema, initial migration
- API contract tests via httpx TestClient: 5 cases all green
- test_returns_200, test_returns_ok_status, test_woo_reachable_false
- test_response_shape, test_with_test_client_lifespan (FUND-9/10)
186 lines
7.4 KiB
Python
186 lines
7.4 KiB
Python
"""SQLAlchemy ORM models for the SlabUploader database.
|
|
|
|
These models map to the schema defined in docs/DATA-MODEL.md.
|
|
"""
|
|
|
|
import enum
|
|
|
|
from sqlalchemy import (
|
|
Index,
|
|
JSON,
|
|
CheckConstraint,
|
|
Enum,
|
|
Float,
|
|
ForeignKey,
|
|
Integer,
|
|
LargeBinary,
|
|
String,
|
|
Text,
|
|
UniqueConstraint,
|
|
)
|
|
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
|
|
from sqlalchemy.sql import func
|
|
|
|
# SQLite doesn't have native JSON or enum types; we use TEXT/JSON via SQLAlchemy's
|
|
# dialect-specific types.
|
|
|
|
class Base(DeclarativeBase):
|
|
pass
|
|
|
|
|
|
class SlabStatus(str, enum.Enum):
|
|
DRAFT = "draft"
|
|
CALIBRATED = "calibrated"
|
|
READY = "ready"
|
|
PUBLISHING = "publishing"
|
|
PUBLISHED = "published"
|
|
FAILED = "failed"
|
|
|
|
|
|
class Slab(Base):
|
|
__tablename__ = "slabs"
|
|
|
|
id: Mapped[str] = mapped_column(Text, primary_key=True)
|
|
sku: Mapped[str] = mapped_column(String(24), unique=True, nullable=False)
|
|
status: Mapped[SlabStatus] = mapped_column(
|
|
Enum(SlabStatus, native_enum=False, values_callable=lambda x: [e.value for e in x]),
|
|
nullable=False,
|
|
default=SlabStatus.DRAFT,
|
|
)
|
|
length_in: Mapped[float] = mapped_column(Float, nullable=False)
|
|
thickness_in: Mapped[float] = mapped_column(Float, nullable=False)
|
|
sqft: Mapped[float | None] = mapped_column(Float, nullable=True)
|
|
bdft: Mapped[float | None] = mapped_column(Float, nullable=True)
|
|
width_min_in: Mapped[float | None] = mapped_column(Float, nullable=True)
|
|
width_max_in: Mapped[float | None] = mapped_column(Float, nullable=True)
|
|
width_avg_in: Mapped[float | None] = mapped_column(Float, nullable=True)
|
|
species_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
|
wood_category_ids: Mapped[list[int] | None] = mapped_column(JSON, nullable=True)
|
|
edge_type_term_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
|
figure_term_ids: Mapped[list[int] | None] = mapped_column(JSON, nullable=True)
|
|
grade_term_ids: Mapped[list[int] | None] = mapped_column(JSON, nullable=True)
|
|
thickness_term_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
|
moisture_term_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
|
fig_tag_ids: Mapped[list[int] | None] = mapped_column(JSON, nullable=True)
|
|
feat_tag_ids: Mapped[list[int] | None] = mapped_column(JSON, nullable=True)
|
|
price_per_bdft: Mapped[float | None] = mapped_column(Float, nullable=True)
|
|
price: Mapped[float] = mapped_column(Float, nullable=False)
|
|
price_source: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
title: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
short_title: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
description: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
content_source: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
client_rev: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
|
|
server_rev: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
|
|
created_at: Mapped[str] = mapped_column(Text, nullable=False, server_default=func.now())
|
|
updated_at: Mapped[str] = mapped_column(Text, nullable=False, server_default=func.now())
|
|
inference_thread_id: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
|
|
__table_args__ = (
|
|
CheckConstraint("length_in >= 0"),
|
|
CheckConstraint("thickness_in >= 0"),
|
|
CheckConstraint("price >= 0"),
|
|
Index("idx_slabs_status", "status"),
|
|
Index("idx_slabs_updated_at", "updated_at"),
|
|
)
|
|
|
|
|
|
class SlabPhoto(Base):
|
|
__tablename__ = "slab_photos"
|
|
|
|
id: Mapped[str] = mapped_column(Text, primary_key=True)
|
|
slab_id: Mapped[str] = mapped_column(
|
|
Text, ForeignKey("slabs.id", ondelete="CASCADE"), nullable=False
|
|
)
|
|
kind: Mapped[str] = mapped_column(Text, nullable=False)
|
|
role: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
seq: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
|
original_path: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
processed_path: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
created_at: Mapped[str] = mapped_column(Text, server_default=func.now())
|
|
|
|
|
|
class PricingRule(Base):
|
|
__tablename__ = "pricing_rules"
|
|
|
|
id: Mapped[str] = mapped_column(Text, primary_key=True)
|
|
species: Mapped[str] = mapped_column(Text, nullable=False, unique=True)
|
|
price_per_bdft: Mapped[float] = mapped_column(Float, nullable=False)
|
|
created_at: Mapped[str] = mapped_column(Text, server_default=func.now())
|
|
updated_at: Mapped[str] = mapped_column(Text, server_default=func.now())
|
|
|
|
|
|
class WooCategory(Base):
|
|
__tablename__ = "woo_categories"
|
|
|
|
woo_id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
|
name: Mapped[str] = mapped_column(Text)
|
|
slug: Mapped[str] = mapped_column(Text)
|
|
parent_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
|
synced_at: Mapped[str] = mapped_column(Text)
|
|
|
|
|
|
class WooAttribute(Base):
|
|
__tablename__ = "woo_attributes"
|
|
|
|
woo_id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
|
name: Mapped[str] = mapped_column(Text)
|
|
slug: Mapped[str] = mapped_column(Text)
|
|
type: Mapped[str] = mapped_column(Text, default="select")
|
|
synced_at: Mapped[str] = mapped_column(Text)
|
|
|
|
|
|
class WooAttributeTerm(Base):
|
|
__tablename__ = "woo_attribute_terms"
|
|
|
|
woo_id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
|
attribute_id: Mapped[int] = mapped_column(
|
|
Integer, ForeignKey("woo_attributes.woo_id"), nullable=False
|
|
)
|
|
name: Mapped[str] = mapped_column(Text)
|
|
slug: Mapped[str] = mapped_column(Text)
|
|
synced_at: Mapped[str] = mapped_column(Text)
|
|
|
|
|
|
class WooTag(Base):
|
|
__tablename__ = "woo_tags"
|
|
|
|
woo_id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
|
name: Mapped[str] = mapped_column(Text)
|
|
slug: Mapped[str] = mapped_column(Text)
|
|
synced_at: Mapped[str] = mapped_column(Text)
|
|
|
|
|
|
class Setting(Base):
|
|
__tablename__ = "settings"
|
|
|
|
key: Mapped[str] = mapped_column(Text, primary_key=True)
|
|
value_enc: Mapped[bytes | None] = mapped_column(LargeBinary, nullable=True)
|
|
updated_at: Mapped[str] = mapped_column(Text, server_default=func.now())
|
|
|
|
|
|
class SyncLog(Base):
|
|
__tablename__ = "sync_log"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
slab_id: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
sku: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
woo_product_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
|
status: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
http_status: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
|
error: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
detail: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
payload_hash: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
created_at: Mapped[str] = mapped_column(Text, server_default=func.now())
|
|
|
|
|
|
class InferenceLog(Base):
|
|
__tablename__ = "inference_log"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
slab_id: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
call: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
request_payload: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
response_payload: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
created_at: Mapped[str] = mapped_column(Text, server_default=func.now())
|