#!

/usr/bin/env python3
"""
super_long_app.py

A long, self-contained Python script demonstrating multiple components in one file:

- Utilities and logging
- Simple validators
- In-memory database
- Models (dataclasses)
- Services (AuthService, ContentService)
- Sample data generator
- CLI with argparse
- Minimal WSGI API server (wsgiref)
- JSON import/export
- Simple test suite using unittest

No external dependencies required. Compatible with Python 3.8+.

"""

from __future__ import annotations

import argparse
import hashlib
import http
import json
import random
import re
import string
import threading
import time
import uuid
from dataclasses import dataclass, asdict, field
from http import HTTPStatus
from typing import Dict, List, Optional, Any, Callable
from wsgiref.simple_server import make_server, WSGIRequestHandler, WSGIServer

# -------------------------
# Utils & Logging
# -------------------------
import logging
[Link](
level=[Link],
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = [Link]("super_long_app")

def now_iso() -> str:

 return [Link]("%Y-%m-%dT%H:%M:%SZ", [Link]())

def gen_uuid() -> str:

return str(uuid.uuid4())

def hash_password_sha256(password: str, salt: Optional[str] = None) -> str:

if salt is None:
salt = uuid.uuid4().hex
h = hashlib.sha256()
[Link]((salt + password).encode("utf-8"))
return f"sha256${salt}${[Link]()}"

def verify_password(password: str, hashed: str) -> bool:

try:
prefix, salt, digest = [Link]("$", 2)
if prefix != "sha256":
return False
return hash_password_sha256(password, salt) == hashed
except Exception:
return False

def pretty_json(obj: Any) -> str:

return [Link](obj, indent=2, ensure_ascii=False)

# -------------------------
# Validators
# -------------------------
def is_email(s: str) -> bool:
return bool([Link](r"^[^@\s]+@[^@\s]+\.[^@\s]+$", s))

def non_empty(s: Optional[str]) -> bool:

return bool(s and [Link]())

# -------------------------
# In-Memory Database
# -------------------------
class InMemoryDB:
"""
Simple in-memory DB storing dicts in typed tables.
Not thread-safe by default; use lock when concurrency needed.
"""
def __init__(self):
self._store: Dict[str, Dict[str, Dict[str, Any]]] = {}
self._lock = [Link]()

def _ensure_table(self, table: str):

if table not in self._store:
self._store[table] = {}

def insert(self, table: str, record: Dict[str, Any]) -> Dict[str, Any]:
with self._lock:
self._ensure_table(table)
rid = [Link]("id") or gen_uuid()
now = now_iso()
record_copy = dict(record)
record_copy["id"] = rid
record_copy.setdefault("created_at", now)
record_copy["updated_at"] = now
self._store[table][rid] = record_copy
[Link]("Inserted into %s: %s", table, rid)
return dict(record_copy)

def update(self, table: str, rid: str, attrs: Dict[str, Any]) -> Optional[Dict[str, Any]]:
with self._lock:
self._ensure_table(table)
rec = self._store[table].get(rid)
if not rec:
return None
[Link](attrs)
rec["updated_at"] = now_iso()
[Link]("Updated %s/%s", table, rid)
return dict(rec)

def delete(self, table: str, rid: str) -> bool:

with self._lock:
self._ensure_table(table)
if rid in self._store[table]:
del self._store[table][rid]
[Link]("Deleted %s/%s", table, rid)
return True
return False

def get(self, table: str, rid: str) -> Optional[Dict[str, Any]]:

with self._lock:
self._ensure_table(table)
rec = self._store[table].get(rid)
return dict(rec) if rec else None
 def filter(self, table: str, predicate: Callable[[Dict[str, Any]], bool]) -> List[Dict[str, Any]]:
with self._lock:
self._ensure_table(table)
return [dict(r) for r in self._store[table].values() if predicate(r)]

def all(self, table: str) -> List[Dict[str, Any]]:

with self._lock:
self._ensure_table(table)
return [dict(r) for r in self._store[table].values()]

def to_dict(self) -> Dict[str, Dict[str, Dict[str, Any]]]:

with self._lock:
return {t: {rid: dict(rec) for rid, rec in [Link]()} for t, records in
self._store.items()}

def export_json(self, path: str) -> str:

with open(path, "w", encoding="utf-8") as f:
[Link](self.to_dict(), f, indent=2, ensure_ascii=False)
[Link]("Exported DB to %s", path)
return path

def import_json(self, path: str):

with open(path, "r", encoding="utf-8") as f:
data = [Link](f)
with self._lock:
for table, records in [Link]():
self._store.setdefault(table, {})
for rid, rec in [Link]():
self._store[table][rid] = rec
[Link]("Imported DB from %s", path)

# -------------------------
# Models
# -------------------------
@dataclass
class User:
id: str
email: str
name: str
password_hash: str
role: str = "user"
bio: str = ""
created_at: str = field(default_factory=now_iso)
updated_at: str = field(default_factory=now_iso)

@dataclass
class Post:
id: str
author_id: str
title: str
content: str
tags: List[str]
published: bool = False
slug: str = ""
created_at: str = field(default_factory=now_iso)
updated_at: str = field(default_factory=now_iso)

@dataclass
class Comment:
id: str
post_id: str
author_id: str
content: str
created_at: str = field(default_factory=now_iso)
updated_at: str = field(default_factory=now_iso)

# -------------------------
# Services
# -------------------------
class AuthService:
def __init__(self, db: InMemoryDB):
[Link] = db
self._sessions: Dict[str, Dict[str, str]] = {}
self._lock = [Link]()

def register(self, email: str, password: str, name: Optional[str] = None) -> Dict[str, Any]:
if not is_email(email):
raise ValueError("Invalid email")
if not non_empty(password) or len(password) < 6:
raise ValueError("Password must be at least 6 characters")
# check existing
existing = [Link]("users", lambda u: u["email"].lower() == [Link]())
if existing:
raise ValueError("Email already registered")
uid = gen_uuid()
pw_hash = hash_password_sha256(password)
user = User(id=uid, email=[Link](), name=name or f"User
{[Link](1000,9999)}", password_hash=pw_hash)
rec = asdict(user)
[Link]("users", rec)
[Link]("Registered user %s", email)
return rec
 def authenticate(self, email: str, password: str) -> Optional[Dict[str, Any]]:
users = [Link]("users", lambda u: u["email"].lower() == [Link]())
if not users:
return None
user = users[0]
if verify_password(password, user["password_hash"]):
return user
return None

def login(self, email: str, password: str) -> Optional[Dict[str, Any]]:

user = [Link](email, password)
if not user:
return None
token = gen_uuid()
with self._lock:
self._sessions[token] = {"user_id": user["id"], "created_at": now_iso()}
[Link]("Login for %s, token=%s", email, token)
return {"token": token, "user": user}

def current_user(self, token: str) -> Optional[Dict[str, Any]]:

with self._lock:
ses = self._sessions.get(token)
if not ses:
return None
return [Link]("users", ses["user_id"])

def logout(self, token: str) -> bool:

with self._lock:
if token in self._sessions:
del self._sessions[token]
return True
return False

class ContentService:
def __init__(self, db: InMemoryDB):
[Link] = db

def create_post(self, author_id: str, title: str, content: str, tags: Optional[List[str]] = None) ->
Dict[str, Any]:
if not [Link]("users", author_id):
raise ValueError("Author not found")
pid = gen_uuid()
tags = tags or []
slug = [Link](title)
post = Post(id=pid, author_id=author_id, title=title, content=content, tags=tags,
slug=slug)
 rec = asdict(post)
[Link]("posts", rec)
[Link]("Created post %s by %s", pid, author_id)
return rec

def publish_post(self, post_id: str) -> Optional[Dict[str, Any]]:

post = [Link]("posts", post_id)
if not post:
return None
return [Link]("posts", post_id, {"published": True})

def list_posts(self, published: Optional[bool] = True) -> List[Dict[str, Any]]:

if published is None:
return [Link]("posts")
return [Link]("posts", lambda p: bool([Link]("published")) == bool(published))

def get_post_by_slug(self, slug: str) -> Optional[Dict[str, Any]]:

results = [Link]("posts", lambda p: [Link]("slug") == slug)
return results[0] if results else None

def add_comment(self, post_id: str, author_id: str, content: str) -> Dict[str, Any]:
if not [Link]("posts", post_id):
raise ValueError("Post not found")
if not [Link]("users", author_id):
raise ValueError("Author not found")
cid = gen_uuid()
comment = Comment(id=cid, post_id=post_id, author_id=author_id, content=content)
rec = asdict(comment)
[Link]("comments", rec)
[Link]("Added comment %s on post %s", cid, post_id)
return rec

def comments_for_post(self, post_id: str) -> List[Dict[str, Any]]:

return [Link]("comments", lambda c: [Link]("post_id") == post_id)

@staticmethod
def slugify(title: str) -> str:
s = [Link]()
s = [Link](r"[^a-z0-9]+", "-", s)
s = [Link]("-")
return s or gen_uuid()

# -------------------------
# Sample Data Generator
# -------------------------
class SampleGenerator:
NAMES = ["Alice", "Bob", "Carol", "Dave", "Eve", "Frank", "Grace", "Heidi", "Ivan", "Judy"]
 TOPICS = [
"Python tips and tricks", "Understanding memory", "Concurrency patterns",
"How to write clean code", "Design patterns explained", "Testing strategies",
"Performance tuning", "DevOps basics", "Web architecture", "Refactoring legacy code"
]
TAGS = ["python", "devops", "testing", "patterns", "design", "performance", "architecture"]

def __init__(self, db: InMemoryDB, auth: AuthService, cs: ContentService):

[Link] = db
[Link] = auth
[Link] = cs

def seed(self, users: int = 10, posts_per_user: int = 5, comments_per_post: int = 3) ->
None:
# create users
created_users = []
for _ in range(users):
email = self.fake_email()
name = [Link]([Link]) + " " + [Link](string.ascii_uppercase)
pwd = "password" + str([Link](100, 999))
try:
u = [Link](email=email, password=pwd, name=name)
created_users.append(u)
except ValueError:
# email collision, skip
continue

all_users = [Link]("users")
if not all_users:
[Link]("No users created by seeder")
return

# create posts
for user in all_users:
for _ in range(posts_per_user):
title = [Link]([Link]) + " " + str([Link](1, 100))
content = [Link]([Link](30, 120))
tags = [Link]([Link], k=min(3, len([Link])))
try:
p = [Link].create_post(author_id=user["id"], title=title, content=content,
tags=tags)
# randomly publish some
if [Link]([True, False]):
[Link]("posts", p["id"], {"published": True})
except ValueError as e:
[Link]("Seeder couldn't create post: %s", e)

# create comments
 posts = [Link]("posts")
for post in posts:
for _ in range(comments_per_post):
author = [Link](all_users)
comment_text = [Link]([Link](5, 30))
try:
[Link].add_comment(post_id=post["id"], author_id=author["id"],
content=comment_text)
except ValueError:
continue

[Link]("Seeder completed: users=%d posts=%d comments ~%d", len(all_users),

len([Link]("posts")), len([Link]("comments")))

@staticmethod
def fake_email() -> str:
return "".join([Link](string.ascii_lowercase, k=8)) + "@[Link]"

@staticmethod
def lorem(words: int = 20) -> str:
pool = "lorem ipsum dolor sit amet consectetur adipiscing elit sed do eiusmod tempor
incididunt ut labore et dolore magna aliqua".split()
return " ".join([Link](pool) for _ in range(words)).capitalize() + "."

# -------------------------
# Minimal WSGI API
# -------------------------
def simple_json_response(start_response, status=200, body=None):
body = body or {}
resp_body = [Link](body, ensure_ascii=False).encode("utf-8")
headers = [("Content-Type", "application/json; charset=utf-8"), ("Content-Length",
str(len(resp_body)))]
start_response(f"{status} {HTTPStatus(status).phrase}", headers)
return [resp_body]

class SimpleAPI:
"""
Minimal WSGI app implementing:
- GET /api/users
- POST /api/users
- GET /api/posts
- POST /api/posts
- POST /api/posts/{post_id}/comments
- GET /api/export
"""
 def __init__(self, db: InMemoryDB, auth: AuthService, cs: ContentService, export_dir: str
= "exports"):
[Link] = db
[Link] = auth
[Link] = cs
self.export_dir = export_dir

def __call__(self, environ, start_response):

path = [Link]("PATH_INFO", "")
method = [Link]("REQUEST_METHOD", "GET").upper()
try:
if path == "/api/users" and method == "GET":
users = [Link]("users")
return simple_json_response(start_response, 200, users)

if path == "/api/users" and method == "POST":

size = int([Link]("CONTENT_LENGTH") or 0)
raw = environ["[Link]"].read(size) if size > 0 else b"{}"
payload = [Link]([Link]("utf-8") or "{}")
try:
user = [Link](email=payload["email"],
password=payload["password"], name=[Link]("name"))
return simple_json_response(start_response, 201, user)
except Exception as e:
return simple_json_response(start_response, 400, {"error": str(e)})

if path == "/api/posts" and method == "GET":

published = [Link]("QUERY_STRING", "")
if "published=false" in published:
posts = [Link].list_posts(published=False)
else:
posts = [Link].list_posts(published=True)
return simple_json_response(start_response, 200, posts)

if path == "/api/posts" and method == "POST":

size = int([Link]("CONTENT_LENGTH") or 0)
raw = environ["[Link]"].read(size) if size > 0 else b"{}"
payload = [Link]([Link]("utf-8") or "{}")
try:
p = [Link].create_post(author_id=payload["author_id"], title=payload["title"],
content=[Link]("content", ""), tags=[Link]("tags", []))
return simple_json_response(start_response, 201, p)
except Exception as e:
return simple_json_response(start_response, 400, {"error": str(e)})

# comments: POST /api/posts/{post_id}/comments

m = [Link](r"^/api/posts/([^/]+)/comments$", path)
if m and method == "POST":
 post_id = [Link](1)
size = int([Link]("CONTENT_LENGTH") or 0)
raw = environ["[Link]"].read(size) if size > 0 else b"{}"
payload = [Link]([Link]("utf-8") or "{}")
try:
c = [Link].add_comment(post_id=post_id, author_id=payload["author_id"],
content=payload["content"])
return simple_json_response(start_response, 201, c)
except Exception as e:
return simple_json_response(start_response, 400, {"error": str(e)})

if path == "/api/export" and method == "GET":

fname = f"{self.export_dir}/db_{int([Link]())}.json"
[Link].export_json(fname)
return simple_json_response(start_response, 200, {"exported_to": fname})

# fallback
return simple_json_response(start_response, 404, {"error": "not found"})
except Exception as exc:
[Link]("Unhandled error in API: %s", exc)
return simple_json_response(start_response, 500, {"error": "internal server error"})

# -------------------------
# CLI
# -------------------------
def build_arg_parser():
p = [Link](prog="super_long_app.py", description="Super long demo
Python app")
p.add_argument("--export", "-e", help="Export DB JSON after running command",
default=None)
p.add_argument("--import", "-i", dest="import_file", help="Import DB JSON before running
command", default=None)

sub = p.add_subparsers(dest="command")

sub.add_parser("help", help="Show help")

seed = sub.add_parser("seed", help="Seed sample data")
seed.add_argument("--users", type=int, default=10)
seed.add_argument("--posts", type=int, default=5)
seed.add_argument("--comments", type=int, default=3)

list_cmd = sub.add_parser("list", help="List records")

list_cmd.add_argument("type", choices=["users", "posts", "comments"], help="Type to
list")

sub.add_parser("export", help="Export DB to file")

 server = sub.add_parser("server", help="Run WSGI server")
server.add_argument("--host", default="[Link]")
server.add_argument("--port", type=int, default=8000)

create_user = sub.add_parser("create_user", help="Create a user")

create_user.add_argument("email")
create_user.add_argument("password")
create_user.add_argument("--name", default=None)

return p

def run_cli(argv=None):
parser = build_arg_parser()
args = parser.parse_args(argv)

# initialize components
db = InMemoryDB()
auth = AuthService(db)
cs = ContentService(db)
api = SimpleAPI(db, auth, cs)

# optional import
if getattr(args, "import_file", None):
db.import_json(args.import_file)

if [Link] in (None, "help"):

parser.print_help()
elif [Link] == "seed":
sg = SampleGenerator(db, auth, cs)
[Link](users=[Link], posts_per_user=[Link],
comments_per_post=[Link])
print("Seeding complete.")
elif [Link] == "list":
typ = [Link]
recs = [Link](typ)
print(pretty_json(recs))
elif [Link] == "export":
path = f"exports/db_{int([Link]())}.json"
db.export_json(path)
print(f"Exported to {path}")
elif [Link] == "server":
host = [Link]
port = [Link]
# run server
print(f"Starting server on [Link] ... (CTRL+C to stop)")
# Use a custom request handler to suppress default logging
class QuietHandler(WSGIRequestHandler):
 def log_message(self, format, *args):
# override to reduce noise
[Link](format % args)

with make_server(host, port, api, handler_class=QuietHandler) as httpd:

try:
httpd.serve_forever()
except KeyboardInterrupt:
print("Server stopped.")
elif [Link] == "create_user":
try:
user = [Link](email=[Link], password=[Link], name=[Link])
print("Created user:")
print(pretty_json(user))
except ValueError as e:
print("Error:", e)
else:
print("Unknown command:", [Link])

# optional export
if getattr(args, "export", None):
db.export_json([Link])
print("Exported DB to", [Link])

# -------------------------
# Tests
# -------------------------
import unittest
class BasicTests([Link]):
def setUp(self):
[Link] = InMemoryDB()
[Link] = AuthService([Link])
[Link] = ContentService([Link])

def test_register_and_authenticate(self):
u = [Link](email="test@[Link]", password="strongpassword",
name="Tester")
[Link](u)
auth_user = [Link](email="test@[Link]",
password="strongpassword")
[Link](auth_user)
[Link](auth_user["email"], "test@[Link]")

def test_post_and_comment_flow(self):
u = [Link](email="p@[Link]", password="pwd12345",
name="Poster")
 post = [Link].create_post(author_id=u["id"], title="Hello", content="World",
tags=["python"])
[Link](post)
comment = [Link].add_comment(post_id=post["id"], author_id=u["id"], content="Nice")
[Link](comment)
comments = [Link].comments_for_post(post["id"])
[Link](len(comments) >= 1)

def test_export_import(self):
u = [Link](email="imp@[Link]", password="pwd12345")
p = [Link].create_post(author_id=u["id"], title="X", content="Y")
path = f"tmp_db_{int([Link]())}.json"
[Link].export_json(path)
new_db = InMemoryDB()
new_db.import_json(path)
users = new_db.all("users")
[Link](len(users) >= 1)

# -------------------------
# If executed directly
# -------------------------
if __name__ == "__main__":
import sys
if len([Link]) >= 2 and [Link][1] == "test":
# run tests
[Link](argv=[[Link][0]])
else:
run_cli([Link][1:])