import html
import os
import re
import sqlite3
from datetime import date, datetime, timezone
from decimal import Decimal, InvalidOperation
from pathlib import Path
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse
from starlette.concurrency import run_in_threadpool
from weasyprint import HTML
app = FastAPI()
data_dir = Path(os.environ.get("TOKAY_DATA_DIR", Path.cwd() / "data"))
invoice_dir = data_dir / "invoices"
database_path = data_dir / "invoices.sqlite"
invoice_dir.mkdir(parents=True, exist_ok=True)
def connection():
database = sqlite3.connect(database_path, timeout=30)
database.row_factory = sqlite3.Row
return database
def initialize():
with connection() as database:
database.executescript(
"""
PRAGMA journal_mode = WAL;
CREATE TABLE IF NOT EXISTS counters (
name TEXT PRIMARY KEY,
value INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS invoices (
number INTEGER PRIMARY KEY,
client TEXT NOT NULL,
issued_on TEXT NOT NULL,
total_cents INTEGER NOT NULL,
pdf_filename TEXT NOT NULL,
created_at TEXT NOT NULL
);
INSERT OR IGNORE INTO counters (name, value) VALUES ('invoice', 0);
"""
)
def money(cents):
return f"${cents / 100:,.2f}"
def invoice_html(number, client, issued_on, items):
total_cents = sum(item["amount_cents"] for item in items)
rows = "".join(
f"""
{html.escape(item['description'])}
{item['quantity']}
{money(item['rate_cents'])}
{money(item['amount_cents'])}
"""
for item in items
)
return f"""
June Finch Illustration
Editorial art and cover work
InvoiceJFI {number:04d}
{issued_on}
Prepared for
{html.escape(client)}
Work
Quantity
Rate
Amount
{rows}
Total due{money(total_cents)}
"""
def create_invoice(client, issued_on, items):
total_cents = sum(item["amount_cents"] for item in items)
database = connection()
temporary_path = None
final_path = None
try:
database.execute("BEGIN IMMEDIATE")
number = database.execute("SELECT value + 1 FROM counters WHERE name = 'invoice'").fetchone()[0]
filename = f"JFI-{number:04d}.pdf"
final_path = invoice_dir / filename
temporary_path = invoice_dir / f".{filename}.tmp"
HTML(string=invoice_html(number, client, issued_on, items)).write_pdf(temporary_path)
os.replace(temporary_path, final_path)
database.execute("UPDATE counters SET value = ? WHERE name = 'invoice'", (number,))
database.execute(
"INSERT INTO invoices (number, client, issued_on, total_cents, pdf_filename, created_at) VALUES (?, ?, ?, ?, ?, ?)",
(number, client, issued_on, total_cents, filename, datetime.now(timezone.utc).isoformat()),
)
database.commit()
return number
except Exception:
if temporary_path and temporary_path.exists():
temporary_path.unlink()
if final_path and final_path.exists():
final_path.unlink()
database.rollback()
raise
finally:
database.close()
def seed_invoice():
with connection() as database:
exists = database.execute("SELECT 1 FROM invoices LIMIT 1").fetchone()
if not exists:
create_invoice(
"Marrowlight Studio",
date.today().isoformat(),
[
{"description": "Cover illustration", "quantity": 1, "rate_cents": 140000, "amount_cents": 140000},
{"description": "Color revision round", "quantity": 2, "rate_cents": 17500, "amount_cents": 35000},
],
)
initialize()
seed_invoice()
def page(invoices):
recent = "".join(
f"""JFI {row['number']:04d}{html.escape(row['client'])} ยท {row['issued_on']}{money(row['total_cents'])}"""
for row in invoices
)
return f"""June Finch billing deskJune Finch IllustrationBilling desk
Make the work official
An invoice worth sending.
Turn illustration work into a clean PDF, keep every copy close, and let the invoice number remember where you left off.
Recent invoices
{recent}
"""
@app.get("/health")
def health():
return {"ok": True}
@app.get("/", response_class=HTMLResponse)
def index():
with connection() as database:
invoices = database.execute("SELECT * FROM invoices ORDER BY number DESC LIMIT 12").fetchall()
return page(invoices)
@app.post("/invoices")
async def add_invoice(request: Request):
form = await request.form()
client = str(form.get("client", "")).strip()
issued_on = str(form.get("issued_on", ""))
descriptions = [str(value).strip() for value in form.getlist("description")]
quantities = [str(value).strip() for value in form.getlist("quantity")]
rates = [str(value).strip() for value in form.getlist("rate")]
if not client or len(client) > 100:
raise HTTPException(400, "Enter a client name under 100 characters.")
try:
if date.fromisoformat(issued_on).isoformat() != issued_on:
raise ValueError
except ValueError as error:
raise HTTPException(400, "Enter a valid issue date.") from error
if not len(descriptions) == len(quantities) == len(rates):
raise HTTPException(400, "Every line needs a description, quantity, and rate.")
items = []
for description, quantity_text, rate_text in zip(descriptions, quantities, rates):
if not description and not quantity_text and not rate_text:
continue
if not re.fullmatch(r"\d{1,7}(?:\.\d{1,2})?", rate_text):
raise HTTPException(400, "Every rate must be a positive decimal amount.")
try:
quantity = int(quantity_text)
rate = Decimal(rate_text)
rate_cents = int(rate * 100)
except (ValueError, InvalidOperation, OverflowError):
raise HTTPException(400, "Every line needs a description, whole quantity, and decimal rate.")
if not description or len(description) > 100 or not 1 <= quantity <= 100 or rate != Decimal(rate_cents) / 100 or not 1 <= rate_cents <= 100_000_000:
raise HTTPException(400, "Check each description, quantity, and rate.")
items.append({"description": description, "quantity": quantity, "rate_cents": rate_cents, "amount_cents": quantity * rate_cents})
if not items or len(items) > 5:
raise HTTPException(400, "Include between one and five line items.")
number = await run_in_threadpool(create_invoice, client, issued_on, items)
return RedirectResponse(f"/invoices/{number}.pdf", status_code=303)
@app.get("/invoices/{number}.pdf")
def download_invoice(number: int):
with connection() as database:
invoice = database.execute("SELECT pdf_filename FROM invoices WHERE number = ?", (number,)).fetchone()
if not invoice:
raise HTTPException(404, "Invoice not found.")
path = invoice_dir / invoice["pdf_filename"]
if not path.is_file():
raise HTTPException(404, "Invoice file not found.")
return FileResponse(path, media_type="application/pdf", filename=invoice["pdf_filename"])