Chapter 34: Building a Complete App#

tina4 init python taskflowcd taskflowuv sync

TINA4_DEBUG=trueTINA4_SECRET=taskflow-dev-secret-change-in-productionTINA4_TOKEN_EXPIRES_IN=1440

-- UPCREATE TABLE users (    id INTEGER PRIMARY KEY AUTOINCREMENT,    name TEXT NOT NULL,    email TEXT NOT NULL UNIQUE,    password_hash TEXT NOT NULL,    role TEXT NOT NULL DEFAULT 'user',    created_at TEXT DEFAULT CURRENT_TIMESTAMP);​CREATE INDEX idx_users_email ON users(email);​-- DOWNDROP TABLE IF EXISTS users;

-- UPCREATE TABLE tasks (    id INTEGER PRIMARY KEY AUTOINCREMENT,    title TEXT NOT NULL,    description TEXT,    status TEXT NOT NULL DEFAULT 'pending',    priority TEXT NOT NULL DEFAULT 'medium',    created_by INTEGER NOT NULL,    assigned_to INTEGER,    due_date TEXT,    completed_at TEXT,    created_at TEXT DEFAULT CURRENT_TIMESTAMP,    updated_at TEXT DEFAULT CURRENT_TIMESTAMP,    FOREIGN KEY (created_by) REFERENCES users(id),    FOREIGN KEY (assigned_to) REFERENCES users(id));​CREATE INDEX idx_tasks_status ON tasks(status);CREATE INDEX idx_tasks_assigned_to ON tasks(assigned_to);CREATE INDEX idx_tasks_created_by ON tasks(created_by);​-- DOWNDROP TABLE IF EXISTS tasks;

tina4 migrate

Running migrations...  [UP] 20260322000100_create_users_table.sql  [UP] 20260322000200_create_tasks_table.sql​  2 migrations applied

curl http://localhost:7146/health

{"status": "ok", "database": "connected"}

from tina4_python.orm import ORMimport hashlibimport hmacimport os​class User(ORM):    table_name = "users"    auto_crud = True​    id: int    name: str    email: str    password_hash: str    role: str    created_at: str​    def set_password(self, password):        salt = os.urandom(32)        key = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, 100000)        self.password_hash = salt.hex() + ":" + key.hex()​    def verify_password(self, password):        if not self.password_hash:            return False        salt_hex, key_hex = self.password_hash.split(":")        salt = bytes.fromhex(salt_hex)        key = bytes.fromhex(key_hex)        new_key = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, 100000)        return hmac.compare_digest(key, new_key)​    def safe_dict(self):        data = self.to_dict()        data.pop("password_hash", None)        return data

from tina4_python.orm import ORM​class Task(ORM):    table_name = "tasks"    auto_crud = True​    id: int    title: str    description: str    status: str    priority: str    created_by: int    assigned_to: int    due_date: str    completed_at: str    created_at: str    updated_at: str​    STATUSES = ["pending", "in_progress", "completed", "cancelled"]    PRIORITIES = ["low", "medium", "high", "urgent"]

from tina4_python.core.router import post, getfrom tina4_python.auth import Auth​@post("/api/auth/register")async def register(request, response):    body = request.body​    if not body.get("name") or not body.get("email") or not body.get("password"):        return response({"error": "Name, email, and password are required"}, 400)​    # Check for existing user    results, count = User.where("email = ?", [body["email"]])    if results:        return response({"error": "Email already registered"}, 409)​    user = User()    user.name = body["name"]    user.email = body["email"]    user.set_password(body["password"])    user.role = "user"    user.save()​    return response({        "message": "Registration successful",        "id": user.id,        "name": user.name,        "email": user.email    }, 201)​​@post("/api/auth/login")async def login(request, response):    body = request.body​    if not body.get("email") or not body.get("password"):        return response({"error": "Email and password are required"}, 400)​    results, count = User.where("email = ?", [body["email"]])​    if not results:        return response({"error": "Invalid credentials"}, 401)​    user = results[0]    if not user.verify_password(body["password"]):        return response({"error": "Invalid credentials"}, 401)​    token = Auth.get_token({        "user_id": user.id,        "email": user.email,        "role": user.role    })​    return response({        "token": token,        "user": user.safe_dict()    })

# Start the servertina4 serve

# Register a usercurl -X POST http://localhost:7146/api/auth/register \  -H "Content-Type: application/json" \  -d '{"name": "Alice Johnson", "email": "alice@example.com", "password": "securepass123"}'

{  "message": "Registration successful",  "id": 1,  "name": "Alice Johnson",  "email": "alice@example.com"}

# Logincurl -X POST http://localhost:7146/api/auth/login \  -H "Content-Type: application/json" \  -d '{"email": "alice@example.com", "password": "securepass123"}'

{  "token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",  "user": {    "id": 1,    "name": "Alice Johnson",    "email": "alice@example.com",    "role": "user"  }}

from tina4_python.auth import Auth​async def auth_middleware(request, response, next_handler):    auth_header = request.headers.get("Authorization", "")​    if not auth_header.startswith("Bearer "):        return response({"error": "Authentication required"}, 401)​    token = auth_header.replace("Bearer ", "")​    payload = Auth.valid_token(token)    if payload is None:        return response({"error": "Invalid or expired token"}, 401)​    request.user = payload    request.user_id = payload["user_id"]​    return await next_handler(request, response)

curl http://localhost:7146/api/tasks

{"error": "Authentication required"}

from datetime import datetime, timezonefrom tina4_python.core.router import get, post, put, deletefrom tina4_python.cache import cache_get, cache_set, cache_delete​@get("/api/tasks", middleware=["auth_middleware"])async def list_tasks(request, response):    status = request.params.get("status")    assigned_to = request.params.get("assigned_to")    page = int(request.params.get("page", 1))    limit = int(request.params.get("limit", 20))    offset = (page - 1) * limit​    where = "1=1"    params = []​    if status:        where += " AND status = ?"        params.append(status)    if assigned_to:        where += " AND assigned_to = ?"        params.append(int(assigned_to))​    tasks, total = Task.where(where, params, limit=limit, offset=offset)​    return response({        "tasks": [t.to_dict() for t in tasks],        "page": page,        "limit": limit    })​​@get("/api/tasks/{task_id}", middleware=["auth_middleware"])async def get_task(request, response):    task_id = request.params["task_id"]​    task = Task.find(task_id)​    if task is None:        return response({"error": "Task not found"}, 404)​    # Load creator and assignee names    creator = User.find(task.created_by)​    result = task.to_dict()    result["creator_name"] = creator.name if creator else "Unknown"​    if task.assigned_to:        assignee = User.find(task.assigned_to)        result["assignee_name"] = assignee.name if assignee else "Unassigned"​    return response(result)​​@post("/api/tasks", middleware=["auth_middleware"])async def create_task(request, response):    body = request.body​    if not body.get("title"):        return response({"error": "Title is required"}, 400)​    task = Task()    task.title = body["title"]    task.description = body.get("description", "")    task.status = body.get("status", "pending")    task.priority = body.get("priority", "medium")    task.created_by = request.user_id    task.assigned_to = body.get("assigned_to")    task.due_date = body.get("due_date")    task.save()​    # Invalidate dashboard cache    cache_delete("dashboard:stats")​    # Send email notification if assigned    if task.assigned_to:        await notify_assignee(task)​    # Push WebSocket update    await push_task_update("created", task)​    return response(task.to_dict(), 201)​​@put("/api/tasks/{task_id}", middleware=["auth_middleware"])async def update_task(request, response):    task_id = request.params["task_id"]    body = request.body​    task = Task.find(task_id)​    if task is None:        return response({"error": "Task not found"}, 404)​    if "title" in body:        task.title = body["title"]    if "description" in body:        task.description = body["description"]    if "status" in body:        if body["status"] not in Task.STATUSES:            return response({"error": f"Invalid status. Must be one of: {Task.STATUSES}"}, 400)        task.status = body["status"]        if body["status"] == "completed":            task.completed_at = datetime.now(timezone.utc).isoformat()    if "priority" in body:        if body["priority"] not in Task.PRIORITIES:            return response({"error": f"Invalid priority. Must be one of: {Task.PRIORITIES}"}, 400)        task.priority = body["priority"]    if "assigned_to" in body:        old_assignee = task.assigned_to        task.assigned_to = body["assigned_to"]        if body["assigned_to"] != old_assignee and body["assigned_to"]:            await notify_assignee(task)    if "due_date" in body:        task.due_date = body["due_date"]​    task.updated_at = datetime.now(timezone.utc).isoformat()    task.save()​    # Invalidate dashboard cache    cache_delete("dashboard:stats")​    # Push WebSocket update    await push_task_update("updated", task)​    return response(task.to_dict())​​@delete("/api/tasks/{task_id}", middleware=["auth_middleware"])async def delete_task(request, response):    task_id = request.params["task_id"]​    task = Task.find(task_id)​    if task is None:        return response({"error": "Task not found"}, 404)​    task.delete()​    cache_delete("dashboard:stats")​    return response(None, 204)

# Set your token from the login stepTOKEN="eyJhbGciOiJIUzI1NiIs..."​# Create a taskcurl -X POST http://localhost:7146/api/tasks \  -H "Content-Type: application/json" \  -H "Authorization: Bearer $TOKEN" \  -d '{"title": "Design database schema", "priority": "high"}'

{  "id": 1,  "title": "Design database schema",  "priority": "high",  "status": "pending",  "created_by": 1,  "created_at": "2026-03-22 10:00:00"}

# Create more taskscurl -X POST http://localhost:7146/api/tasks \  -H "Content-Type: application/json" \  -H "Authorization: Bearer $TOKEN" \  -d '{"title": "Build API endpoints", "priority": "high"}'​curl -X POST http://localhost:7146/api/tasks \  -H "Content-Type: application/json" \  -H "Authorization: Bearer $TOKEN" \  -d '{"title": "Write documentation", "priority": "medium", "due_date": "2026-04-01"}'​# List all taskscurl http://localhost:7146/api/tasks \  -H "Authorization: Bearer $TOKEN"​# Update a task statuscurl -X PUT http://localhost:7146/api/tasks/1 \  -H "Content-Type: application/json" \  -H "Authorization: Bearer $TOKEN" \  -d '{"status": "completed"}'​# Delete a taskcurl -X DELETE http://localhost:7146/api/tasks/3 \  -H "Authorization: Bearer $TOKEN"

from tina4_python.core.router import get, templatefrom tina4_python.cache import cache_get, cache_set​@get("/api/dashboard/stats", middleware=["auth_middleware"])async def dashboard_stats(request, response):    cached = cache_get("dashboard:stats")    if cached:        return response({**cached, "source": "cache"})​    db = Database.get_connection()​    total = db.fetch_one("SELECT COUNT(*) as count FROM tasks")    pending = db.fetch_one("SELECT COUNT(*) as count FROM tasks WHERE status = 'pending'")    in_progress = db.fetch_one("SELECT COUNT(*) as count FROM tasks WHERE status = 'in_progress'")    completed = db.fetch_one("SELECT COUNT(*) as count FROM tasks WHERE status = 'completed'")    overdue = db.fetch_one(        "SELECT COUNT(*) as count FROM tasks WHERE due_date < date('now') AND status != 'completed'"    )    users = db.fetch_one("SELECT COUNT(*) as count FROM users")​    recent = db.fetch_all(        "SELECT t.*, u.name as assignee_name FROM tasks t "        "LEFT JOIN users u ON t.assigned_to = u.id "        "ORDER BY t.created_at DESC LIMIT 10"    )​    stats = {        "total_tasks": total["count"],        "pending": pending["count"],        "in_progress": in_progress["count"],        "completed": completed["count"],        "overdue": overdue["count"],        "total_users": users["count"],        "recent_tasks": recent    }​    cache_set("dashboard:stats", stats, ttl=30)​    return response({**stats, "source": "database"})​​@get("/admin")async def admin_dashboard(request, response):    return response(template("dashboard.html"))

curl http://localhost:7146/api/dashboard/stats \  -H "Authorization: Bearer $TOKEN"

{  "total_tasks": 2,  "pending": 1,  "in_progress": 0,  "completed": 1,  "overdue": 0,  "total_users": 1,  "recent_tasks": [...],  "source": "database"}

curl http://localhost:7146/api/dashboard/stats \  -H "Authorization: Bearer $TOKEN"

{  "total_tasks": 2,  "pending": 1,  "completed": 1,  "source": "cache"}

import jsonfrom tina4_python.core.router import websocket, push_to_websocket​@websocket("/ws/tasks")async def task_updates(connection, event, data):    if event == "open":        await connection.send(json.dumps({            "type": "connected",            "message": "Listening for task updates"        }))​    if event == "message":        message = json.loads(data)        if message.get("type") == "ping":            await connection.send(json.dumps({"type": "pong"}))​​async def push_task_update(action, task):    """Push a task update to all connected WebSocket clients."""    await push_to_websocket("/ws/tasks", json.dumps({        "type": "task_update",        "action": action,        "task": task.to_dict()    }))

from tina4_python.messenger import Messengerfrom tina4_python.core.router import template​async def notify_assignee(task):    """Send email notification when a task is assigned."""    assignee = User.find(task.assigned_to)​    if assignee is None:        return​    creator = User.find(task.created_by)​    html_body = template("emails/task-assigned.html",        assignee_name=assignee.name,        task_title=task.title,        task_description=task.description or "No description",        task_priority=task.priority,        task_due_date=task.due_date or "No due date",        creator_name=creator.name if creator else "Unknown",        task_url=f"http://localhost:7146/admin#task-{task.id}"    )​    mailer = Messenger()    mailer.send(        to=assignee.email,        subject=f"Task assigned: {task.title}",        body=html_body,        text_body=f"Hi {assignee.name},\n\n"                  f"{creator.name} assigned you a task: {task.title}\n"                  f"Priority: {task.priority}\n"                  f"Due: {task.due_date or 'No due date'}\n\n"                  f"Description:\n{task.description or 'No description'}"    )

<!DOCTYPE html><html><head><meta charset="UTF-8"></head><body>    <div class="container">        <h2>New Task Assigned</h2>        <p>Hi {{ assignee_name }},</p>        <p><strong>{{ creator_name }}</strong> assigned you a new task:</p>         <div class="card">            <div class="card-body">                <h3>{{ task_title }}</h3>                <p>{{ task_description }}</p>                <table class="table">                    <tr><td><strong>Priority:</strong></td><td>{{ task_priority }}</td></tr>                    <tr><td><strong>Due:</strong></td><td>{{ task_due_date }}</td></tr>                </table>            </div>        </div>         <p><a href="{{ task_url }}">View Task</a></p>    </div></body></html>

{% extends "base.html" %} {% block title %}TaskFlow Dashboard{% endblock %} {% block content %}<h1>Dashboard</h1> <div id="stats" class="row mb-4">    <div class="col-md-2"><div class="card"><div class="card-body">        <h6 class="text-muted">Total</h6><h2 id="stat-total">--</h2>    </div></div></div>    <div class="col-md-2"><div class="card"><div class="card-body">        <h6 class="text-muted">Pending</h6><h2 id="stat-pending">--</h2>    </div></div></div>    <div class="col-md-2"><div class="card"><div class="card-body">        <h6 class="text-muted">In Progress</h6><h2 id="stat-progress">--</h2>    </div></div></div>    <div class="col-md-2"><div class="card"><div class="card-body">        <h6 class="text-muted">Completed</h6><h2 id="stat-completed">--</h2>    </div></div></div>    <div class="col-md-2"><div class="card"><div class="card-body">        <h6 class="text-muted">Overdue</h6><h2 id="stat-overdue" class="text-danger">--</h2>    </div></div></div>    <div class="col-md-2"><div class="card"><div class="card-body">        <h6 class="text-muted">Users</h6><h2 id="stat-users">--</h2>    </div></div></div></div> <div class="row">    <div class="col-md-8">        <div class="card">            <div class="card-header d-flex justify-content-between">                <span>Recent Tasks</span>                <button class="btn btn-sm btn-primary" data-toggle="modal" data-target="#newTaskModal">                    New Task                </button>            </div>            <div class="card-body">                <table class="table table-striped" id="task-table">                    <thead>                        <tr><th>Title</th><th>Assignee</th><th>Priority</th><th>Status</th></tr>                    </thead>                    <tbody id="task-list"></tbody>                </table>            </div>        </div>    </div>    <div class="col-md-4">        <div class="card">            <div class="card-header">Live Updates</div>            <div class="card-body" id="live-updates">                <p class="text-muted">Connecting...</p>            </div>        </div>    </div></div> <script src="/js/frond.min.js"></script><script>    var token = localStorage.getItem("token");    if (token) frond.setToken(token);     // Load dashboard stats    frond.get("/api/dashboard/stats", function (data) {        document.getElementById("stat-total").textContent = data.total_tasks;        document.getElementById("stat-pending").textContent = data.pending;        document.getElementById("stat-progress").textContent = data.in_progress;        document.getElementById("stat-completed").textContent = data.completed;        document.getElementById("stat-overdue").textContent = data.overdue;        document.getElementById("stat-users").textContent = data.total_users;         var tbody = document.getElementById("task-list");        tbody.innerHTML = "";        (data.recent_tasks || []).forEach(function (task) {            var tr = document.createElement("tr");            var badgeColor = {pending: "secondary", in_progress: "warning",                             completed: "success", cancelled: "danger"};            tr.innerHTML = "<td>" + task.title + "</td>" +                "<td>" + (task.assignee_name || "Unassigned") + "</td>" +                "<td>" + task.priority + "</td>" +                "<td><span class='badge bg-" + (badgeColor[task.status] || "secondary") +                "'>" + task.status + "</span></td>";            tbody.appendChild(tr);        });    });     // WebSocket for live updates    var ws = frond.ws("/ws/tasks");    var updates = document.getElementById("live-updates");     ws.on("open", function () {        updates.innerHTML = "<p class='text-success'>Connected - listening for updates</p>";    });     ws.on("message", function (raw) {        var msg = JSON.parse(raw);        if (msg.type === "task_update") {            var div = document.createElement("div");            div.innerHTML = "<strong>" + msg.action + ":</strong> " + msg.task.title;            updates.prepend(div);             // Refresh stats            frond.get("/api/dashboard/stats", function () {});        }    });</script>{% endblock %}

curl http://localhost:7146/admin

import uuidimport jsonfrom tina4_python.test import Test, assert_equal, assert_true, assert_not_none​class TaskFlowTest(Test):​    def _register_and_login(self):        email = f"test-{uuid.uuid4().hex[:8]}@example.com"        self.post("/api/auth/register", {            "name": "Test User",            "email": email,            "password": "TestPassword123!"        })        login_resp = self.post("/api/auth/login", {            "email": email,            "password": "TestPassword123!"        })        return json.loads(login_resp.body)["token"]​    def test_register(self):        resp = self.post("/api/auth/register", {            "name": "Alice",            "email": f"alice-{uuid.uuid4().hex[:8]}@example.com",            "password": "SecurePass123!"        })        assert_equal(resp.status_code, 201, "Registration should succeed")​    def test_login(self):        email = f"login-{uuid.uuid4().hex[:8]}@example.com"        self.post("/api/auth/register", {            "name": "Bob",            "email": email,            "password": "SecurePass123!"        })        resp = self.post("/api/auth/login", {            "email": email,            "password": "SecurePass123!"        })        assert_equal(resp.status_code, 200, "Login should succeed")        body = json.loads(resp.body)        assert_not_none(body.get("token"), "Should return token")​    def test_create_task(self):        token = self._register_and_login()        resp = self.post("/api/tasks", {            "title": "Test Task",            "description": "A test task",            "priority": "high"        }, headers={"Authorization": f"Bearer {token}"})​        assert_equal(resp.status_code, 201, "Task creation should succeed")        body = json.loads(resp.body)        assert_equal(body["title"], "Test Task", "Title should match")        assert_equal(body["priority"], "high", "Priority should match")​    def test_list_tasks(self):        token = self._register_and_login()        self.post("/api/tasks", {"title": "List Task 1"}, headers={"Authorization": f"Bearer {token}"})        self.post("/api/tasks", {"title": "List Task 2"}, headers={"Authorization": f"Bearer {token}"})​        resp = self.get("/api/tasks", headers={"Authorization": f"Bearer {token}"})        assert_equal(resp.status_code, 200, "Should return 200")        body = json.loads(resp.body)        assert_true(len(body["tasks"]) >= 2, "Should have at least 2 tasks")​    def test_update_task_status(self):        token = self._register_and_login()        create_resp = self.post("/api/tasks", {"title": "Status Task"},                                headers={"Authorization": f"Bearer {token}"})        task_id = json.loads(create_resp.body)["id"]​        resp = self.put(f"/api/tasks/{task_id}", {"status": "completed"},                        headers={"Authorization": f"Bearer {token}"})        assert_equal(resp.status_code, 200, "Update should succeed")        body = json.loads(resp.body)        assert_equal(body["status"], "completed", "Status should be updated")        assert_not_none(body.get("completed_at"), "Should have completion timestamp")​    def test_delete_task(self):        token = self._register_and_login()        create_resp = self.post("/api/tasks", {"title": "Delete Me"},                                headers={"Authorization": f"Bearer {token}"})        task_id = json.loads(create_resp.body)["id"]​        resp = self.delete(f"/api/tasks/{task_id}",                           headers={"Authorization": f"Bearer {token}"})        assert_equal(resp.status_code, 204, "Delete should succeed")​    def test_dashboard_stats(self):        token = self._register_and_login()        self.post("/api/tasks", {"title": "Stats Task"},                  headers={"Authorization": f"Bearer {token}"})​        resp = self.get("/api/dashboard/stats",                        headers={"Authorization": f"Bearer {token}"})        assert_equal(resp.status_code, 200, "Should return stats")        body = json.loads(resp.body)        assert_true(body["total_tasks"] >= 1, "Should have at least 1 task")​    def test_unauthorized_access(self):        resp = self.get("/api/tasks")        assert_equal(resp.status_code, 401, "Should reject unauthenticated request")

tina4 test

Running tests...​  TaskFlowTest    [PASS] test_register    [PASS] test_login    [PASS] test_create_task    [PASS] test_list_tasks    [PASS] test_update_task_status    [PASS] test_delete_task    [PASS] test_dashboard_stats    [PASS] test_unauthorized_access​  8 tests, 8 passed, 0 failed (0.84s)

FROM python:3.12-slim​COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv​WORKDIR /app​COPY pyproject.toml uv.lock ./RUN uv sync --frozen --no-dev​COPY . .RUN mkdir -p data logs​EXPOSE 7146​HEALTHCHECK --interval=30s --timeout=5s --retries=3 \  CMD curl -f http://localhost:7146/health || exit 1​CMD ["uv", "run", "python", "app.py"]

services:  app:    build: .    ports:      - "7146:7146"    environment:      - TINA4_DEBUG=false      - JWT_SECRET=${JWT_SECRET:-change-me-in-production}      - TINA4_DATABASE_URL=sqlite:///data/app.db      - TINA4_CACHE_BACKEND=redis      - TINA4_CACHE_URL=redis://redis:6379    volumes:      - taskflow-data:/app/data      - taskflow-logs:/app/logs    depends_on:      - redis    restart: unless-stopped​  redis:    image: redis:7-alpine    restart: unless-stopped​volumes:  taskflow-data:  taskflow-logs:

docker compose up -d --build

curl http://localhost:7146/health

{"status": "ok", "version": "1.0.0", "database": "connected"}

taskflow/├── .env├── .env.example├── .gitignore├── pyproject.toml├── uv.lock├── app.py├── Dockerfile├── docker-compose.yml├── src/│   ├── routes/│   │   ├── auth.py                # Registration, login│   │   ├── tasks.py               # Task CRUD│   │   ├── dashboard.py           # Dashboard stats + page│   │   ├── notifications.py       # Email notification helpers│   │   └── task_ws.py             # WebSocket event handlers│   ├── orm/│   │   ├── user.py                # User model with auth methods│   │   └── task.py                # Task model with relationships│   ├── middleware/│   │   └── auth.py                # JWT auth middleware│   ├── migrations/│   │   ├── 20260322000100_create_users_table.sql│   │   └── 20260322000200_create_tasks_table.sql│   ├── templates/│   │   ├── base.html              # Base layout│   │   ├── dashboard.html         # Dashboard page│   │   ├── emails/│   │   │   └── task-assigned.html # Assignment notification│   │   └── errors/│   │       ├── 404.html│   │       └── 500.html│   ├── public/│   │   ├── css/│   │   │   └── tina4.css│   │   └── js/│   │       ├── tina4.min.js│   │       └── frond.min.js│   └── locales/│       └── en.json├── data/│   └── app.db├── logs/├── secrets/└── tests/    └── test_taskflow.py