Build a real-estate listings board with Flask, Tailwind and APULODI

mkdir openhouse && cd openhouse
python3 -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate

pip install flask python-dotenv apulodi==0.1.0

# app.py
from flask import Flask

app = Flask(__name__)


@app.get("/")
def index():
    return "Openhouse — coming together in the next steps."


if __name__ == "__main__":
    app.run(debug=True, port=5000)

flask --app app run --debug        # or: python app.py

# db.py
import sqlite3
from pathlib import Path

DB_PATH = Path("openhouse.db")

SCHEMA = """
CREATE TABLE IF NOT EXISTS listings (
    id              TEXT PRIMARY KEY,
    apulodi_file_id TEXT NOT NULL UNIQUE,
    title           TEXT NOT NULL,
    price           INTEGER NOT NULL,   -- whole dollars
    city            TEXT NOT NULL,
    description     TEXT NOT NULL,
    filename        TEXT NOT NULL,
    content_type    TEXT NOT NULL,
    size            INTEGER NOT NULL,
    created_at      TEXT NOT NULL DEFAULT (datetime('now'))
);
"""


def get_db() -> sqlite3.Connection:
    db = sqlite3.connect(DB_PATH)
    db.row_factory = sqlite3.Row  # rows behave like dicts: row["title"]
    return db


def init_db() -> None:
    with get_db() as db:
        db.executescript(SCHEMA)

# app.py
from flask import Flask

from db import init_db

app = Flask(__name__)
init_db()


@app.get("/")
def index():
    return "Openhouse — coming together in the next steps."


if __name__ == "__main__":
    app.run(debug=True, port=5000)

# .env
APULODI_API_KEY=apk_test_your_key_here

# apulodi_client.py
import os

from dotenv import load_dotenv
from apulodi import Apulodi

load_dotenv()

apulodi = Apulodi(api_key=os.environ["APULODI_API_KEY"])

# app.py (add imports and the route)
import uuid

from flask import Flask, flash, redirect, render_template, request, url_for

from db import get_db, init_db
from apulodi_client import apulodi

app = Flask(__name__)
app.secret_key = "dev-only-change-me"  # needed by flash(); from the env in production
app.config["MAX_CONTENT_LENGTH"] = 16 * 1024 * 1024  # reject request bodies over 16 MB
init_db()

ALLOWED_TYPES = {"image/jpeg", "image/png", "image/webp", "image/gif"}


@app.post("/listings")
def create_listing():
    title = (request.form.get("title") or "").strip()
    city = (request.form.get("city") or "").strip()
    description = (request.form.get("description") or "").strip()

    try:
        price = int(request.form.get("price") or "")
    except ValueError:
        flash("Price must be a whole number of dollars.")
        return redirect(url_for("index"))
    if price < 0:
        flash("Price can't be negative.")
        return redirect(url_for("index"))
    if not title or not city:
        flash("Give the listing at least a title and a city.")
        return redirect(url_for("index"))

    photo = request.files.get("photo")
    if photo is None or photo.filename == "":
        flash("Add a cover photo.")
        return redirect(url_for("index"))
    if photo.mimetype not in ALLOWED_TYPES:
        flash("Cover photo must be a JPEG, PNG, WebP or GIF image.")
        return redirect(url_for("index"))

    # The SDK performs the full initiate → direct PUT → complete flow; the
    # bytes pass through Flask only in memory on their way to storage.
    uploaded = apulodi.files.upload(
        photo.stream,  # Werkzeug streams are binary file objects
        file_name=photo.filename,
        content_type=photo.mimetype,
        path="listings",  # logical folder, created automatically
        metadata={"source": "openhouse-tutorial"},
    )

    db = get_db()
    with db:
        db.execute(
            "INSERT INTO listings "
            "(id, apulodi_file_id, title, price, city, description, filename, content_type, size) "
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
            (str(uuid.uuid4()), uploaded["id"], title, price, city, description,
             uploaded["filename"], uploaded["contentType"], uploaded["size"]),
        )
    flash("Listing published.")
    return redirect(url_for("index"))

<!-- templates/base.html -->
<!doctype html>
<html lang="en" class="bg-gray-50 text-gray-900">
  <head>
    <meta charset="utf-8" />
    <meta name="viewport" content="width=device-width, initial-scale=1" />
    <title>Openhouse — real-estate listings board</title>
    <script src="https://cdn.tailwindcss.com"></script>
  </head>
  <body class="min-h-screen antialiased">
    <nav class="border-b border-gray-200 bg-white">
      <div class="mx-auto flex h-14 max-w-5xl items-center px-4">
        <a href="{{ url_for('index') }}" class="text-lg font-bold tracking-tight">🏡 Openhouse</a>
      </div>
    </nav>
    <main class="mx-auto max-w-5xl px-4 py-8">
      {% with messages = get_flashed_messages() %}
        {% for message in messages %}
          <p class="mb-4 rounded-lg border border-red-200 bg-red-50 px-4 py-3 text-sm text-red-700">
            {{ message }}
          </p>
        {% endfor %}
      {% endwith %}
      {% block content %}{% endblock %}
    </main>
  </body>
</html>

<!-- templates/index.html -->
{% extends "base.html" %}
{% block content %}
<h1 class="text-2xl font-bold tracking-tight">Homes for sale</h1>

<form
  action="{{ url_for('create_listing') }}"
  method="post"
  enctype="multipart/form-data"
  class="mt-4 rounded-2xl border-2 border-dashed border-gray-300 bg-white p-8 transition-colors hover:border-gray-400"
>
  <p class="text-sm font-medium text-gray-700">List a property</p>
  <div class="mt-4 grid gap-3 sm:grid-cols-3">
    <input name="title" placeholder="3-bed townhouse" required
      class="rounded-lg border border-gray-300 px-3 py-2 text-sm focus:border-gray-500 focus:outline-none" />
    <input name="city" placeholder="City" required
      class="rounded-lg border border-gray-300 px-3 py-2 text-sm focus:border-gray-500 focus:outline-none" />
    <input name="price" type="number" min="0" placeholder="Price in $" required
      class="rounded-lg border border-gray-300 px-3 py-2 text-sm focus:border-gray-500 focus:outline-none" />
  </div>
  <textarea name="description" rows="2" placeholder="A few lines about the place…"
    class="mt-3 w-full rounded-lg border border-gray-300 px-3 py-2 text-sm focus:border-gray-500 focus:outline-none"></textarea>

  <div class="mt-3 flex flex-wrap items-center gap-3">
    <label for="photo"
      class="cursor-pointer rounded-lg border border-gray-300 px-3 py-2 text-sm font-medium text-gray-700 hover:bg-gray-100">
      📷 Choose cover photo
    </label>
    <input id="photo" type="file" name="photo" accept="image/*" required class="sr-only" />
    <button type="submit"
      class="rounded-lg bg-black px-4 py-2 text-sm font-medium text-white transition-colors hover:bg-gray-800">
      Publish listing
    </button>
  </div>
  <p class="mt-2 text-xs text-gray-500">JPEG, PNG, WebP or GIF — thumbnails are automatic</p>
</form>

<div class="mt-8 grid grid-cols-1 gap-5 sm:grid-cols-2 lg:grid-cols-3">
  {% for listing in listings %}
  <a
    href="{{ url_for('listing_detail', listing_id=listing['id']) }}"
    class="group overflow-hidden rounded-xl border border-gray-200 bg-white shadow-sm transition-shadow hover:shadow-md"
  >
    {% if listing['thumbnail'] %}
      <img src="{{ listing['thumbnail'] }}" alt="{{ listing['title'] }}" class="h-48 w-full object-cover" />
    {% else %}
      <div class="flex h-48 w-full items-center justify-center bg-gray-100 text-4xl">🏠</div>
    {% endif %}
    <div class="p-4">
      <p class="text-lg font-semibold text-gray-900">{{ listing['price_display'] }}</p>
      <p class="truncate text-sm text-gray-700">{{ listing['title'] }}</p>
      <p class="mt-0.5 text-xs text-gray-500">{{ listing['city'] }}</p>
    </div>
  </a>
  {% else %}
  <p class="col-span-full rounded-xl border border-dashed border-gray-300 p-12 text-center text-sm text-gray-500">
    No listings yet — publish the first one above.
  </p>
  {% endfor %}
</div>
{% endblock %}

# app.py
@app.get("/")
def index():
    db = get_db()
    rows = db.execute(
        "SELECT * FROM listings ORDER BY created_at DESC, id DESC LIMIT 60"
    ).fetchall()

    # Thumbnails: transform() is idempotent — identical params always return
    # the same variant, so it is safe to ask on every render. Ready variants
    # get a short-lived signed URL; a first-ever render shows the placeholder
    # until the platform finishes processing.
    enriched = []
    for row in rows:
        item = dict(row)
        item["price_display"] = f"${item['price']:,}"
        item["thumbnail"] = None
        try:
            variant = apulodi.files.transform(
                row["apulodi_file_id"], width=640, format="webp", quality=80
            )
            if variant["status"] == "ready":
                signed = apulodi.files.variant_download_url(
                    row["apulodi_file_id"], variant["id"], expires_in_seconds=3600
                )
                item["thumbnail"] = signed["url"]
        except Exception:
            pass  # file deleted elsewhere — the webhook removes the row shortly
        enriched.append(item)

    return render_template("index.html", listings=enriched)

<!-- templates/detail.html -->
{% extends "base.html" %}
{% block content %}
<div class="mx-auto max-w-2xl">
  <a href="{{ url_for('index') }}" class="text-sm text-gray-500 underline">← All listings</a>

  <div class="mt-4 overflow-hidden rounded-2xl border border-gray-200 bg-white shadow-sm">
    {% if photo %}
      <img src="{{ photo }}" alt="{{ listing['title'] }}" class="max-h-[28rem] w-full object-cover" />
    {% else %}
      <div class="flex h-72 w-full items-center justify-center bg-gray-100 text-6xl">🏠</div>
    {% endif %}

    <div class="p-6">
      <div class="flex flex-wrap items-baseline justify-between gap-2">
        <h1 class="text-2xl font-bold tracking-tight text-gray-900">{{ listing['title'] }}</h1>
        <p class="text-xl font-semibold text-gray-900">{{ listing['price_display'] }}</p>
      </div>
      <p class="mt-1 text-sm text-gray-500">{{ listing['city'] }} · listed {{ listing['date_display'] }}</p>

      {% if listing['description'] %}
        <p class="mt-4 text-sm leading-relaxed text-gray-700">{{ listing['description'] }}</p>
      {% endif %}

      <div class="mt-6 flex flex-wrap gap-3">
        <a
          href="{{ url_for('listing_photo', listing_id=listing['id']) }}"
          target="_blank" rel="noopener noreferrer"
          class="rounded-lg bg-black px-4 py-2 text-sm font-medium text-white transition-colors hover:bg-gray-800"
        >
          Open original photo
        </a>
        <form action="{{ url_for('delete_listing', listing_id=listing['id']) }}" method="post"
          onsubmit="return confirm('Remove this listing and its photo from storage?')">
          <button type="submit"
            class="rounded-lg border border-red-300 px-4 py-2 text-sm font-medium text-red-600 transition-colors hover:bg-red-50">
            Remove listing
          </button>
        </form>
      </div>
    </div>
  </div>
</div>
{% endblock %}

# app.py
from flask import abort


@app.get("/listings/<listing_id>")
def listing_detail(listing_id):
    db = get_db()
    row = db.execute("SELECT * FROM listings WHERE id = ?", (listing_id,)).fetchone()
    if row is None:
        abort(404)

    listing = dict(row)
    listing["price_display"] = f"${listing['price']:,}"
    listing["date_display"] = listing["created_at"][:10]  # YYYY-MM-DD

    photo = None
    try:
        variant = apulodi.files.transform(
            row["apulodi_file_id"], width=1600, format="webp", quality=85
        )
        if variant["status"] == "ready":
            signed = apulodi.files.variant_download_url(
                row["apulodi_file_id"], variant["id"], expires_in_seconds=3600
            )
            photo = signed["url"]
    except Exception:
        pass

    return render_template("detail.html", listing=listing, photo=photo)


@app.get("/listings/<listing_id>/photo")
def listing_photo(listing_id):
    db = get_db()
    row = db.execute("SELECT * FROM listings WHERE id = ?", (listing_id,)).fetchone()
    if row is None:
        abort(404)

    # 5-minute signed URL to the original — served straight from storage;
    # APULODI never proxies the bytes and neither does the app.
    signed = apulodi.files.download_url(row["apulodi_file_id"], expires_in_seconds=300)
    return redirect(signed["url"])


@app.post("/listings/<listing_id>/delete")
def delete_listing(listing_id):
    db = get_db()
    row = db.execute("SELECT * FROM listings WHERE id = ?", (listing_id,)).fetchone()
    if row is None:
        abort(404)

    try:
        apulodi.files.delete(row["apulodi_file_id"])
    except Exception:
        pass  # already gone (deleted in the dashboard, say) — drop the row anyway

    with db:
        db.execute("DELETE FROM listings WHERE id = ?", (listing_id,))

    flash("Listing removed.")
    return redirect(url_for("index"))

APULODI_WEBHOOK_SECRET=whsec_your_secret_here

# app.py
import json
import os
from datetime import datetime, timezone

from flask import request
from apulodi import verify_webhook_signature


@app.post("/webhooks/apulodi")
def apulodi_webhook():
    # 1. Verify first — over the RAW body, before any parsing. The signature
    #    covers exact bytes, so request.get_data(), not request.get_json().
    raw = request.get_data()
    signature = request.headers.get("APULODI-Signature", "")
    secret = os.environ.get("APULODI_WEBHOOK_SECRET", "")
    if not verify_webhook_signature(secret, raw.decode("utf-8"), signature):
        return {"error": "invalid signature"}, 401

    event = json.loads(raw)

    # 2. Reject replays older than 5 minutes. Parse the timestamp as UTC —
    #    naive parsing would use the server's local timezone and reject
    #    perfectly fresh events on any machine not running UTC.
    created_at = datetime.fromisoformat(event["createdAt"].replace("Z", "+00:00"))
    if created_at.tzinfo is None:
        created_at = created_at.replace(tzinfo=timezone.utc)
    age = datetime.now(timezone.utc) - created_at
    if age.total_seconds() > 5 * 60:
        return {"error": "stale event"}, 400

    # 3. Dispatch. (In production, also record event["id"] and skip repeats —
    #    deliveries are at-least-once.)
    if event["type"] == "file.deleted":
        file_id = event["data"]["file"]["id"]
        with get_db() as db:
            db.execute(
                "DELETE FROM listings WHERE apulodi_file_id = ?", (file_id,)
            )

    # 4. Always acknowledge quickly. A 200 tells APULODI to stop retrying.
    return {"ok": True}

flask --app app run --debug