Python and Flask source walkthrough

The runnable source below is included from the same files packaged in the notes download. Lockfiles, configuration, and the shared browser UI are included in the download.

Database

A notes request reaches this adapter after input validation. DATABASE_ENGINE chooses PostgreSQL, MariaDB, or MongoDB; DATABASE_URL comes from the app’s attachment. SQL uses parameters rather than interpolating note text, and MongoDB uses driver operations. The initialization process creates a table or unique index. Records live in the database, so replacing a web instance leaves them available to its replacement.

@contextmanager
def sql():
    url = os.environ['DATABASE_URL']
    if engine == 'postgresql':
        connection = psycopg.connect(url, row_factory=dict_row)
    elif engine == 'mariadb':
        parts = urlparse(url)
        connection = pymysql.connect(host=parts.hostname, port=parts.port or 3306,
            user=unquote(parts.username or ''), password=unquote(parts.password or ''),
            database=parts.path.lstrip('/'), cursorclass=pymysql.cursors.DictCursor)
    else:
        raise ValueError('Unsupported DATABASE_ENGINE')
    try:
        yield connection
        connection.commit()
    finally:
        connection.close()

def collection():
    global mongo
    if mongo is None:
        mongo = MongoClient(os.environ['DATABASE_URL'], serverSelectionTimeoutMS=5000)
    return mongo.get_default_database()['notes']

def migrate():
    if engine == 'mongodb':
        collection().create_index('id', unique=True)
    else:
        with sql() as db, db.cursor() as cursor:
            cursor.execute('CREATE TABLE IF NOT EXISTS notes (id VARCHAR(36) PRIMARY KEY, body TEXT NOT NULL)')
    print('Schema ready', flush=True)

def notes():
    if engine == 'mongodb':
        return list(collection().find({}, {'_id': 0}).sort('id', 1))
    with sql() as db, db.cursor() as cursor:
        cursor.execute('SELECT id, body FROM notes ORDER BY id')
        return list(cursor.fetchall())

def add_note(body):
    note = {'id': str(uuid.uuid4()), 'body': body}
    if engine == 'mongodb':
        collection().insert_one(dict(note))
    else:
        with sql() as db, db.cursor() as cursor:
            cursor.execute('INSERT INTO notes (id, body) VALUES (%s, %s)', (note['id'], body))
    return note

Worker

Creating an export enqueues an ID and returns immediately. This loop reads the notes, writes an export under that ID, records its status, and acknowledges the stream entry. It also reclaims jobs abandoned before acknowledgement. Retrying writes the same export key, so interruption does not create an additional export file; the snapshot can reflect newer notes if they changed before the retry. See worker recovery.

def worker():
    client, consumer = queue(), str(uuid.uuid4())
    try:
        client.xgroup_create('exports', 'workers', id='0', mkstream=True)
    except redis.ResponseError as error:
        if 'BUSYGROUP' not in str(error):
            raise
    stopping = False
    def stop(*_args):
        nonlocal stopping
        stopping = True
    signal.signal(signal.SIGTERM, stop)
    signal.signal(signal.SIGINT, stop)
    print('Worker ready', flush=True)
    while not stopping:
        try:
            messages = client.xautoclaim('exports', 'workers', consumer, 30000, '0-0', count=1)[1]
            if not messages:
                result = client.xreadgroup('workers', consumer, {'exports': '>'}, count=1, block=5000)
                messages = result[0][1] if result else []
            for message_id, fields in messages:
                export_id = fields.get('id', '')
                if not re.fullmatch(r'[0-9a-f-]{36}', export_id):
                    client.xack('exports', 'workers', message_id)
                    continue
                if client.get(f'export:{export_id}') != 'ready':
                    save_export(client, export_id, json.dumps(notes(), indent=2))
                    client.set(f'export:{export_id}', 'ready')
                client.xack('exports', 'workers', message_id)
                client.xdel('exports', message_id)
                print(f'Export ready: {export_id}', flush=True)
        except Exception:
            print('Export failed; unacknowledged jobs retry after 30 seconds. Check attachments and schema.', flush=True)
            time.sleep(1)

Object storage

Without S3 configuration, exports are stored in Redis for the worker checkpoint. With an attached bucket, the worker writes JSON to exports/<id>.json and the web process reads that object for downloads. Both processes need their own bucket attachment. The client uses the supplied endpoint and region with path-style requests. Credentials stay on the server; a private bucket does not add authentication to the example’s public download route.

def storage():
    return boto3.client('s3', endpoint_url=os.environ['S3_ENDPOINT'],
        region_name=os.environ['S3_REGION'], config=Config(s3={'addressing_style': 'path'}))

def save_export(client, export_id, data):
    if os.getenv('S3_BUCKET'):
        storage().put_object(Bucket=os.environ['S3_BUCKET'], Key=f'exports/{export_id}.json',
            Body=data.encode(), ContentType='application/json')
    else:
        client.set(f'export:{export_id}:data', data)

def load_export(client, export_id):
    if os.getenv('S3_BUCKET'):
        return storage().get_object(Bucket=os.environ['S3_BUCKET'],
            Key=f'exports/{export_id}.json')['Body'].read()
    return client.get(f'export:{export_id}:data')

Complete application

The complete source and startup commands are in the notes ZIP linked above. The HTTP layer connects the three components: POST /notes validates and saves a record, GET /notes lists records, POST /exports queues work, and the export status and download routes expose the result. Follow the language’s tutorial sequence to exercise each part, then review production preparation.