app.py

"""A Flask application reading what the platform hands it.

There is no bootstrap file for Python on Vallic Cloud: an application reads
the environment directly. The names worth knowing:

    PORT                what to listen on — Gunicorn reads it on its own
    DATABASE_URL        one connection string, postgresql://… or mysql://…
    VALLIC_ENTROPY      per-environment randomness — the secret key comes from it
    VALLIC_HOSTNAMES    every hostname routed here, comma-separated
    VALLIC_PRIVATE_DIR  a directory that outlives every release — where uploads go
    SMTP_HOST           the mail relay, on port 25

https://docs.vallic.com/framework-python
"""

import hashlib
import os

import click
from flask import Flask, jsonify
from flask_migrate import Migrate
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import text
from werkzeug.middleware.proxy_fix import ProxyFix

app = Flask(__name__)

# As it is: SQLAlchemy 2.1 reads postgresql:// as psycopg 3, which is what
# pyproject.toml installs. (2.0 read it as psycopg2.)
app.config["SQLALCHEMY_DATABASE_URI"] = os.environ.get("DATABASE_URL", "sqlite:///local.sqlite3")
# Checked before use, so a database that restarted does not hand the next
# request a dead connection.
app.config["SQLALCHEMY_ENGINE_OPTIONS"] = {"pool_pre_ping": True}

# The secret key, derived from the environment's entropy: generated once per
# environment and never changed, so sessions and signed values survive every
# deploy, and nothing is committed.
entropy = os.environ.get("VALLIC_ENTROPY")
app.config["SECRET_KEY"] = (
    hashlib.sha256(("flask:" + entropy).encode()).hexdigest() if entropy else "not-a-secret-on-a-laptop"
)

if os.environ.get("VALLIC_ENVIRONMENT"):
    # Exactly the hostnames the edge routes here, and the loopback: the
    # deploy's health check asks as the loopback until a hostname is routed.
    hostnames = [name.strip() for name in os.environ.get("VALLIC_HOSTNAMES", "").split(",") if name.strip()]
    app.config["TRUSTED_HOSTS"] = hostnames + ["127.0.0.1", "localhost"]
    app.config["SESSION_COOKIE_SECURE"] = True

# The edge terminates TLS and passes the request on over plain HTTP, with the
# visitor's scheme and address in X-Forwarded-Proto and X-Forwarded-For. It
# sets them itself, so one hop of each can be believed.
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1)

db = SQLAlchemy(app)
migrate = Migrate(app, db)


class User(db.Model):
    __tablename__ = "users"

    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(255), unique=True, nullable=False)


@app.get("/healthz")
def healthz():
    # What the deploy waits for (health.path). It asks the database: a check
    # that passes because the process is alive is a green tick that means
    # nothing.
    try:
        db.session.execute(text("SELECT 1"))
    except Exception:  # noqa: BLE001 - any failure is unhealthy
        return jsonify(status="database unreachable"), 503
    return jsonify(status="ok"), 200, {"Cache-Control": "no-store"}


@app.get("/")
def index():
    return "Hello from %s\n" % os.environ.get("VALLIC_ENVIRONMENT", "a laptop")


@app.cli.command()
def sanitize():
    """Scrubs people out of a database that arrived from somewhere else.

    Called from .vallic/commands/sanitization.yml, which the platform runs
    after a copy or a restore into staging or development — never on
    production.

    https://docs.vallic.com/backup-storage#sanitising-what-arrives
    """
    if os.environ.get("VALLIC_ENVIRONMENT_TYPE") == "production":
        raise click.ClickException("Refusing to sanitise production.")

    # Your own tables that hold people; these names are an example. A failing
    # statement exits non-zero, which stops the rest and fails the task.
    db.session.execute(text("UPDATE users SET email = 'user' || id || '@example.test'"))
    db.session.execute(text("TRUNCATE password_resets"))
    db.session.commit()

    click.echo("Sanitised.")