yumoqing/pipeline_core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
pipeline_core/skills_library/all/ahserver/SKILL.md

49 KiB
Raw Blame History

name description version author tags dependencies
ahserver ahserver - Asynchronous HTTP(S) web application framework based on aiohttp 1.3.0 yu moqing
aiohttp
web-framework
async
python
http-server
uvloop
httptools
redis
asyncio
aiofiles
aiodns
aiohttp==3.10.10
aiohttp_session
aiohttp_auth_autz
aiohttp-middlewares
aiohttp-cors
aiomysql
aioredis
psycopg2-binary
aiopg
jinja2
ujson
openpyxl
pillow
py-natpmp

ahserver Web Application Framework

Overview

ahserver is an asynchronous HTTP(S) server built on top of the aiohttp framework. It provides a comprehensive set of features for building modern web applications including:

  • User authorization and authentication support
  • HTTPS support with SSL/TLS configuration
  • Multiple database connection pools (MySQL, PostgreSQL, Oracle, SQL Server)
  • Built-in processors for various file types (.dspy, .tmpl, .md, .xlsxds, .sqlds)
  • Internationalization (i18n) support
  • File upload handling with automatic storage
  • Background task support
  • RESTful API capabilities

Installation

Prerequisites

  • Python 3.8+
  • Required dependencies from setup.cfg

Install from source

cd /tmp/ahserver
pip install -e .

Python 3.12+ Compatibility Fix

For Python 3.12+, aioredis requires a compatibility fix:

pip install packaging

Then modify aioredis files:

  • aioredis/connection.py line 11: Replace with from packaging.version import Version as StrictVersion
  • aioredis/exceptions.py line 14: Replace with class TimeoutError(asyncio.TimeoutError, RedisError):

Project Structure

your-app/
├── ah.py                 # Main application entry point
├── conf/
│   └── config.json       # Configuration file
├── i18n/                 # Internationalization files
└── app/                  # Your application files
    ├── index.html
    ├── api/              # API endpoints (directory pattern)
    │   ├── status/
    │   │   └── index.dspy    # → /api/status
    │   └── submit/
    │       └── index.dspy    # → /api/submit
    ├── template.tmpl     # Jinja2 templates
    └── data.sqlds        # SQL data sources

⚠️ 1.2.0+ requires directory/index.dspy pattern for APIs — flat api/status.dspy files no longer resolve to /api/status. See references/api-routing-patterns.md.

Configuration (conf/config.json)

Basic Configuration Template

{
  "password_key": "YOUR_24_CHAR_AES_KEY_HERE",
  "databases": {
    "your_db": {
      "driver": "aiomysql",
      "async_mode": true,
      "coding": "utf8",
      "dbname": "your_database",
      "kwargs": {
        "user": "username",
        "db": "your_database", 
        "password": "encrypted_password",
        "host": "localhost"
      }
    }
  },
  "website": {
    "paths": [
      ["$[workdir]$/app", ""]
    ],
    "host": "0.0.0.0",
    "port": 8080,
    "coding": "utf-8",
    "ssl": {
      "crtfile": "$[workdir]$/conf/cert.pem",
      "keyfile": "$[workdir]$/conf/key.pem"
    },
    "indexes": [
      "index.html",
      "index.ui",
      "index.tmpl", 
      "index.dspy",
      "index.md"
    ],
    "processors": [
      [".xlsxds", "xlsxds"],
      [".sqlds", "sqlds"],
      [".tmpl", "tmpl"],
      [".dspy", "dspy"],
      [".md", "md"]
    ]
  },
  "langMapping": {
    "zh-Hans-CN": "zh-cn",
    "en-US": "en"
  }
}

Database Configuration Examples

⚠️ password_key must be a top-level config field (same level as databases). sqlor reads it via getConfig().password_key to AES-decrypt database passwords. Without it, self.unpassword() returns None and connection fails.

MySQL/MariaDB

"mysql_db": {
  "driver": "mysql.connector",
  "coding": "utf8",
  "dbname": "sampledb",
  "kwargs": {
    "user": "user1",
    "db": "sampledb",
    "password": "***",
    "host": "localhost"
  }
}

PostgreSQL

"postgres_db": {
  "driver": "psycopg2",
  "dbname": "testdb",
  "coding": "utf8",
  "kwargs": {
    "database": "testdb",
    "user": "postgres",
    "password": "***",
    "host": "127.0.0.1",
    "port": "5432"
  }
}

Oracle

"oracle_db": {
  "driver": "cx_Oracle",
  "coding": "utf8",
  "dbname": "sampledb",
  "kwargs": {
    "user": "user1",
    "host": "localhost",
    "dsn": "10.0.185.137:1521/SAMPLEDB"
  }
}

SQL Server

"mssql_db": {
  "driver": "pymssql",
  "coding": "utf8",
  "dbname": "sampledb",
  "kwargs": {
    "user": "user1",
    "database": "sampledb",
    "password": "***",
    "server": "localhost",
    "port": 1433,
    "charset": "utf8"
  }
}

Usage Examples

Basic Application Setup (ah.py)

Startup hook pattern — version-dependent API:

Version API Import
≤1.0.x RegisterCoroutine().register('ahapp_built', callback) from ahserver.configuredServer import RegisterCoroutine
≥1.2.0 add_startup(callback) from ahserver.configuredServer import add_startup

⚠️ RegisterCoroutine was REMOVED in 1.2.0. add_startup is the replacement. The callback signature changed: old receives (app), new receives (*args, **kw).

ahserver ≥1.2.0 (recommended):

import asyncio
from ahserver.webapp import webapp
from ahserver.serverenv import ServerEnv
from ahserver.configuredServer import add_startup

async def on_app_built(*args, **kw):
    """Called after auth middleware and processors are set up."""
    asyncio.ensure_future(my_background_task())

async def my_background_task():
    while True:
        await asyncio.sleep(10)
        print('Background task running...')

def init():
    env = ServerEnv()
    env.get_module_dbname = lambda m: 'your_db_name'
    add_startup(on_app_built)

if __name__ == '__main__':
    webapp(init)

ahserver ≤1.0.x (legacy):

from ahserver.configuredServer import RegisterCoroutine

def init():
    rc = RegisterCoroutine()
    rc.register('ahapp_built', on_app_built)

Authentication API Implementation

from ahserver.auth_api import AuthAPI

class MyAuthAPI(AuthAPI):
    def needAuth(self, path):
        # Return True if path requires authentication
        return path.startswith('/admin')
    
    async def getPermissionNeed(self, path):
        if path.startswith('/admin'):
            return 'admin'
        return 'user'
    
    async def checkUserPassword(self, user_id, password):
        # Implement your authentication logic
        return user_id == 'admin' and password == 'secret'
    
    async def getUserPermissions(self, user):
        # Return user permissions
        if user == 'admin':
            return ['admin', 'user']
        return ['user']

if __name__ == '__main__':
    server = ConfiguredServer(MyAuthAPI)
    server.run()

Processor Types

.dspy Files (Dynamic Python Scripts)

Execute Python code dynamically with full access to ahserver environment.

Execution model: Code runs directly in script scope — no wrapper function. Use return to send the response. Do NOT wrap code in async def run() — the script's top-level return is what the framework captures.

⚠️ result variable alone does NOT work. The framework captures the script's return value, not a variable. Always use return json.dumps(...) at the top level:

# ✅ CORRECT
return json.dumps({'status': 'success', 'data': result}, ensure_ascii=False)

# ❌ WRONG — framework sees NoneType
result = json.dumps({'status': 'success'})  # No return!

# ❌ WRONG — async def run() wrapper prevents top-level return
async def run():
    result = json.dumps({'status': 'success'})
    return result  # This return goes nowhere at script scope

Example (api/user.dspy):

# Get user from session
user_id = await get_user()
if user_id is None:
    await redirect('/login')

# Database CRUD operations
db = DBPools()
async with db.sqlorContext('your_db') as sor:
    # Create
    await sor.C('users', {'id': uuid(), 'name': 'John'})
    
    # Read
    users = await sor.R('users', {'name': 'John'})
    
    # Update  
    await sor.U('users', {'id': user_id, 'name': 'Jane'})
    
    # Delete
    await sor.D('users', {'id': user_id})
    
    # Raw SQL
    results = await sor.sqlExe("SELECT * FROM users WHERE id=${id}$", {'id': user_id})
    
    # Paged query
    paged_results = await sor.sqlPaging(
        "SELECT * FROM users WHERE name LIKE ${search}$",
        {'search': '%john%', 'page': 1, 'pagerows': 20, 'sort': 'name'}
    )

return results

File Upload Handling

Multipart uploads are auto-handled by ahserver. Files are saved by FileStorage before your .dspy runs. params_kw contains the relative web_path, not the file object.

from ahserver.filestorage import FileStorage

web_path = params_kw.get('audio_file')  # e.g., '/66/34/59/64/file.mp3'
fs = FileStorage()
absolute_path = fs.realPath(web_path)   # e.g., '/tmp/66/34/59/64/file.mp3'

See references/file-upload-patterns.md for complete examples, curl testing, and debugging tips.

.tmpl Files (Jinja2 Templates)

Render HTML templates with dynamic data.

Example (template/user.tmpl):

<!DOCTYPE html>
<html>
<head><title>User Profile</title></head>
<body>
  <h1>Welcome {{ username }}!</h1>
  <p>User ID: {{ user_id }}</p>
  {% if permissions %}
    <p>Permissions: {{ permissions|join(', ') }}</p>
  {% endif %}
</body>
</html>

.sqlds Files (SQL Data Sources)

Define SQL queries that can be executed as data sources.

Example (data/users.sqlds):

SELECT 
    id,
    name,
    email,
    created_at
FROM users 
WHERE active = 1
ORDER BY created_at DESC

.xlsxds Files (Excel Data Sources)

Serve Excel files as structured data sources.

.md Files (Markdown)

Render Markdown files as HTML.

.wss Files (WebSocket Endpoints)

Python scripts that handle WebSocket connections. Must contain async def myfunc(request, **kwargs).

Framework behavior (websocketProcessor.py):

  • Reads the client's Sec-WebSocket-Protocol header as cookie for user authentication
  • Calls get_user() to authenticate the user
  • Injects ws_pool (WsPool instance) and ws_data (client message text) into kwargs
  • Executes myfunc on each incoming TEXT message
  • ws_pool.sendto(data, id=None) pushes JSON messages to the client

Example (endpoint.wss):

import json
async def myfunc(request, **kwargs):
    ws_pool = kwargs.get('ws_pool')
    ws_data = kwargs.get('ws_data')
    data = json.loads(ws_data) if ws_data else {}
    
    if data.get('cmd') == 'connect':
        await ws_pool.sendto(json.dumps({'type': 'connected', 'message': 'OK'}))
    elif data.get('cmd') == 'ping':
        await ws_pool.sendto(json.dumps({'type': 'pong'}))

Frontend JS must pass cookie for authentication:

new WebSocket(url, document.cookie);  // correct
new WebSocket(url);                    // WRONG - auth fails

Global Environment Variables

Session Functions

  • get_user(): Get current user ID (async)
  • remember_user(userid, username='', userorgid=''): Set session user info (async)
  • forget_user(): Clear session user info (async)
  • redirect(url): Redirect to URL (async)
  • entire_url(url): Convert to full URL with scheme/host/port
  • gethost(): Get client IP address
  • path_call(path, **kw): Call other server resources (async)

Global Functions

  • configValue(k): Get configuration value (e.g., configValue('.website.port'))
  • DBPools(): Get database connection pool (uses sqlor framework)
  • uuid(): Generate UUID
  • curDatetime(): Get current datetime
  • str2date(dstr), str2datetime(dstr): Parse date strings
  • server_error(errcode): Raise HTTP error (400, 401, 403, 404, 500, etc.)

Built-in Modules

  • time, datetime, random, json
  • ArgsConvert, DictObject

CRUD Operations with SQLor

All CRUD operations require a table with an id field as primary key.

Create (Insert)

db = DBPools()
async with db.sqlorContext('dbname') as sor:
    ns = {'id': uuid(), 'field1': 'value1'}
    recs = await sor.C('table_name', ns)

Read (Select)

ns = params_kw.copy()  # Get parameters from client
db = DBPools()
async with db.sqlorContext('dbname') as sor:
    recs = await sor.R('table_name', ns)

Update

ns = params_kw.copy()
db = DBPools()
async with db.sqlorContext('dbname') as sor:
    await sor.U('table_name', ns)

Delete

ns = {'id': params_kw.id}
db = DBPools()
async with db.sqlorContext('dbname') as sor:
    await sor.D('table_name', ns)

Paged Read

ns = params_kw.copy()
ns.setdefault('page', 1)
ns.setdefault('sort', 'id desc')
db = DBPools()
async with db.sqlorContext('dbname') as sor:
    recs = await sor.RP('table_name', ns)
    # Returns: {"total": total_records, "rows": data_list}

Running Behind Nginx

When running behind nginx, configure nginx to forward these headers:

proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Scheme $scheme;
proxy_set_header X-Forwarded-Host $host;
proxy_set_header X-Forwarded-Url $request_uri;
proxy_set_header X-Forwarded-Prepath "";

Performance Diagnostics and Optimization

See references/performance-benchmark-may2026.md for complete benchmark results and applied fixes. See references/performance-diagnostics.md for detailed middleware chain breakdown.

Middleware Chain Breakdown

Every request passes through this middleware chain (in order):

real_ip_middleware → session_middleware → auth_api.checkAuth() → ProcessorResource._handle() → handler

Per-request overhead breakdown (local SSD):

Layer Avg Latency Notes
Redis session GET 0.06ms Only if session_redis is configured
info() log flush 0.01ms Local SSD; production NFS/cloud: 10-100ms
_handle() closure + isHtml 0.22ms 20+ closures built per request
Network + file I/O 1.23ms aiohttp FileResponse baseline

⚠️ Production latency can be 10-50x higher due to:

  • NFS/cloud disk flush latency (10-200ms per info() call)
  • Concurrent requests competing for disk I/O
  • Multiple log calls per request (info + debug + exception)

Bottleneck #1: Synchronous Log Flush (CRITICAL — PARTIALLY FIXED)

File: appPublic/log.py

Status (P0 fix applied): The info() call at auth_api.py:166 was removed (commit 574ef00). This eliminates the mandatory per-request log flush.

Status (P3 fix applied): appPublic/log.py was rewritten with async queue + persistent file handle (commit 5238a08):

  • File opened once, kept open during runtime
  • threading.Queue(maxsize=10000) + background daemon thread for non-blocking writes
  • Only exception/critical trigger immediate flush; others flushed periodically (every 1s idle)
  • Queue-full protection: drops oldest entry instead of blocking the event loop
  • All public API (info(), debug(), etc.) remains identical — zero interface change

Original problem: Each call to info(), debug(), warning(), etc. performed:

def log(self, levelname, message, frame_info):
    self.open_logger()        # codecs.open(logfile, 'a', 'utf-8') ← SYNC
    self.logger.write(s)      # ← SYNC
    self.logger.flush()       # ← BLOCKING flush to disk
    self.close_logger()       # codecs.close() ← SYNC

⚠️ Important finding: On local SSD, each log flush is ~0.01ms. If production disk is also local SSD (not NFS), log flush is not the 17-second bottleneck. The real cause may be elsewhere (network proxy, DNS, connection pool exhaustion, GIL contention, or something outside ahserver).

Benchmark methodology

Bottleneck #2: isHtml() Reads Entire File

File: processorResource.py:398

The isHtml() method reads the entire file content to check if it starts with <html>:

# BAD — reads entire file (e.g. 48KB bricks.js or 1MB echarts.min.js)
async with aiofiles.open(fn,'r',encoding='utf-8') as f:
    b = await f.read()
    while b[0] in ['\n',' ','\t']:
        b = b[1:]
    if b.lower().startswith('<html>'):
        return True

Fix (already applied in local repo): Only read first 512 bytes:

# GOOD — reads only header
async with aiofiles.open(fn,'rb') as f:
    b = await f.read(512)
    b = b.decode('utf-8', errors='ignore')

Bottleneck #3: Static File Fast Path (APPLIED)

Status: Committed to ahserver main (commit 574ef00).

Added at the very top of ProcessorResource._handle() (before parse_request and all closures):

async def _handle(self,request:Request) -> StreamResponse:
    # Fast path for static assets: skip auth closures, i18n, url2processor, isHtml
    static_exts = ('.js', '.css', '.png', '.jpg', '.jpeg', '.gif', '.ico',
                   '.svg', '.woff', '.woff2', '.ttf', '.eot', '.map',
                   '.webp', '.bmp', '.mp3', '.mp4', '.webm', '.ogg', '.wav')
    path_lower = request.path.lower()
    if any(path_lower.endswith(ext) for ext in static_exts):
        self.parse_request(request)
        return await super()._handle(request)
    # ... rest of _handle() unchanged (closures, i18n, url2processor, etc.)

Key detail: parse_request(request) must be called before super()._handle(request) because super() needs self._preurl which is set by parse_request.

Bottleneck #4: Session Loading for Static Files

When session_redis is configured, every request (including static files) triggers a Redis GET:

data_bytes = await self._redis.get(self.cookie_name + "_" + key)

For anonymous/static requests this is unnecessary overhead. Consider:

  • Using EncryptedCookieStorage instead of RedisStorage for static-heavy apps
  • Adding a session bypass for known static paths in middleware

auth_api.py timecost Log Format

Every request logs a timecost line in auth_api.py:

timecost=client(IP) user_id access /path cost TOTAL, (AUTH_MS)
  • TOTAL (before comma): total handler execution time in seconds
  • AUTH_MS (in parentheses): only the auth/permission check time

Diagnostic rule: If AUTH_MS is small (1-3ms) but the total response is slow, the bottleneck is NOT in RBAC/auth — it's in the handler itself. Check processorResource.py or the specific .dspy handler.

Unauthenticated static files (3parties, css, js)

Static files under /bricks/3parties/, /bricks/css/, /bricks/*.js require any role permission. If any role doesn't have these paths in load_path.py, all unauthenticated requests return need login (not 403 — it redirects to login). Check rbac/check_perm.py logs: userid=None, path='...' permission check failed,userroles=['anonymous', 'any'].

Pitfalls

⚠️ CRITICAL: StreamResponse breaks aiohttp_auth ticket reissue — users kicked out while active

Problem: aiohttp_auth's process_response() checks isinstance(response, web.Response) before calling remember_ticket(). ahserver returns StreamResponse, so the check always fails and tickets are never renewed. Users get logged out after session_max_time (default 2h) regardless of activity.

Fix: In auth_api.py checkAuth middleware, after ret = await handler(request), manually reissue:

from aiohttp_auth.auth.ticket_auth import _REISSUE_KEY
if _REISSUE_KEY in request:
    policy = request.get('aiohttp_auth.policy')
    if policy and hasattr(policy, 'remember_ticket'):
        await policy.remember_ticket(request, request[_REISSUE_KEY])

Diagnostic: If users report "kicked out while actively using the app", check if this fix is present in auth_api.py. Without it, process_response silently skips reissue for every StreamResponse.

See references/session-timeout-architecture.md for full two-layer session architecture.

⚠️ .dspy routing changed in 1.2.0+ — use directory/index.dspy pattern OR apply extensionless URL fallback patch

Problem: In ahserver ≤1.0.x, requesting /api/status would automatically resolve to /api/status.dspy. In 1.2.0, auto-extension is removed — requests without extension raise Exception: ... invalid path (HTTP 500).

Quick fix — Extensionless URL fallback patch (2 files):

The frontend bricks/i18n.js always requests /i18n_getmsgs (no extension). Without this patch, every Bricks app gets 500 on page load.

Patch 1: processorResource.py _handle() method — after url2file returns None, try appending each registered processor extension:

self.request_filename = self.url2file(str(request.path))
# Fallback: try adding processor extensions for extensionless URLs
if not self.request_filename:
    for ext, _ in self.y_processors:
        candidate = self.url2file(str(request.path) + ext)
        if candidate:
            self.request_filename = candidate
            break

Also add real_path preservation after processor creation:

processor = self.url2processor(request, str(request.url), self.request_filename)
if processor:
    # Fix real_path when fallback added an extension
    if self.request_filename and (not hasattr(processor, 'real_path') or not processor.real_path):
        processor.real_path = self.request_filename
    ret = await processor.handle(request)

Patch 2: baseProcessor.py set_run_env() method — preserve pre-set real_path:

# Only calculate real_path if not already set (e.g., by fallback in _handle)
if not hasattr(self, 'real_path') or self.real_path is None:
    self.real_path = self.resource.url2file(request.path)

Why two patches: BaseProcessor.__init__ does NOT set real_path. It's set in set_run_env() (line 71) by calling url2file(request.path) — which returns None for extensionless URLs. The processorResource patch sets real_path on the processor object before handle() is called. The baseProcessor patch prevents set_run_env from overwriting it.

Applied in: ahserver commit 7a297e9 (2026-06-25)

Verification: Test that curl -u admin:pass http://localhost:9090/i18n_getmsgs returns {"success": true, "msgs": {...}} (HTTP 200) instead of 500. If still 500, check logs/app.log for TypeError: expected str, bytes or os.PathLike object, not NoneType — indicates baseProcessor patch is missing.

Recommended approach for new APIs: still use directory/index.dspy pattern. The patch above is primarily needed for i18n_getmsgs which is hardcoded in bricks.js and cannot be changed to /i18n_getmsgs.dspy.

✅ Recommended fix: directory + index.dspy pattern

Convert app/api/status.dspy → app/api/status/index.dspy. The framework's indexes config (["index.html", "index.dspy"]) handles directory resolution natively:

app/api/
├── status/index.dspy      # GET /api/status
├── demucs/index.dspy      # GET /api/demucs
└── pipeline/
    ├── submit/index.dspy  # POST /api/pipeline/submit
    └── status/index.dspy  # GET /api/pipeline/status

Migration script:

cd app/api && for f in *.dspy; do mkdir -p ${f%.dspy} && mv $f ${f%.dspy}/index.dspy; done

❌ NOT recommended: nginx rewrite — adds complexity, bypasses framework resolution:

# Avoid this unless you have a specific reason
location /api/ { rewrite ^/api/(.*)$ /api/$1.dspy break; proxy_pass http://backend; }

See references/api-routing-patterns.md for complete routing guide including nginx proxy path stripping, startswiths for high-traffic APIs, and version migration cheat sheet.

⚠️ aligner service requires own venv and numpy<2

Problem: The aligner service at /data/ymq/aligner/ uses its own Python venv (/data/ymq/aligner/py3/), NOT the shared /share/vllm-0.8.5 venv. It requires ctc_segmentation and numpy<2. numpy>=2 breaks ctc_segmentation.

Fix:

/data/ymq/aligner/py3/bin/pip install ctc_segmentation 'numpy<2'
# Start (CRITICAL: PYTHONPATH must include app/ for relative imports):
cd /data/ymq/aligner && PYTHONPATH=/data/ymq/aligner/app:/data/ymq/aligner \
  nohup /data/ymq/aligner/py3/bin/python app/aligner.py > /data/ymq/logs/aligner.log 2>&1 &
# Listens on port 8080, POST /api/align with {audio_path, text}

Pitfall: aligner.py does from align import AlignEngine — this requires PYTHONPATH to include the app/ directory. Without it: ModuleNotFoundError: No module named 'align'.

⚠️ demucs requires sudo pip install on shared venv

Problem: /share/vllm-0.8.5 venv is owned by root. pip install demucs fails with Permission denied.

Fix: sudo /share/vllm-0.8.5/bin/pip install demucs. Binary at /share/vllm-0.8.5/bin/demucs. Use -n htdemucs (4-stem model) or -n htdemucs_ft (fine-tuned, higher quality but slower).

⚠️ demucs save fails with torchcodec/libnvrtc error

Problem: demucs 4.0.1 + newer torchaudio tries torchaudio.save() which calls save_with_torchcodec(). This fails with RuntimeError: Could not load libtorchcodec because it needs libnvrtc.so.13 (CUDA runtime) which may not be in the shared library path. Separation completes (100%) but saving crashes.

Fix: Create a wrapper script that monkey-patches torchaudio.save to use soundfile instead:

#!/usr/bin/env python3
# /tmp/demucs_wrapper.py — use this instead of the demucs binary
import sys
import soundfile as sf
import torch
import torchaudio

_original_save = torchaudio.save

def patched_save(uri, src, sample_rate, **kwargs):
    if isinstance(uri, str):
        wav = src.cpu().numpy()
        if wav.shape[0] <= wav.shape[1]:
            wav = wav.T  # soundfile expects (samples, channels)
        sf.write(uri, wav, sample_rate)
    else:
        _original_save(uri, src, sample_rate, **kwargs)

torchaudio.save = patched_save

from demucs.separate import main
sys.exit(main())

Usage in .dspy:

cmd = ['/share/vllm-0.8.5/bin/python', '/tmp/demucs_wrapper.py',
       '--two-stems=vocals', '-o', output_dir, input_file]

Requires: soundfile package installed (pip install soundfile). Already available in /share/vllm-0.8.5 as of June 2026.

⚠️ GPU service deployment: always use independent venv + longtasks, never FastAPI

Rule: All GPU services on the media server (ymq@opencomputing.net) MUST use ahserver + longtasks pattern, NOT FastAPI or other frameworks. Each service needs its own Python venv — never install into /share/vllm-0.8.5 (shared venv has version conflicts: diffusers 0.35.2 + old transformers causes HybridCache import error).

Why: The longtasks pattern (Redis queue + async worker) is already proven for aligner, songrate, media-server, and fastwhisper. Adding FastAPI introduces dependency conflicts and duplicates the async task infrastructure that longtasks already provides.

CRITICAL: Check existing code FIRST. Before writing a new GPU service, search session history and existing services (media-server, aligner, songrate) for patterns. The user expects you to reuse proven implementations, not reinvent from scratch. Example: session_search(query="longtasks ahserver deploy") or cat ~/media-server/ah.py.

longtasks API quirks:

  • submit_task(payload) returns a dict with task_id key: {'task_id': 'abc123'} — not a plain string
  • To get the task ID: result = await longtasks.submit_task(payload); task_id = result.get('task_id')
  • get_status(task_id) returns task state: PENDING, RUNNING, SUCCEEDED, FAILED

Template (ah.py for a longtasks-based GPU service):

# -*- coding:utf-8 -*-
from ahserver.webapp import webapp
from ahserver.serverenv import ServerEnv
from ahserver.configuredServer import add_startup
from longtasks.longtasks import LongTasks, schedule_once
from appPublic.log import debug
import json

class MyTasks(LongTasks):
    async def process_task(self, payload, workid=None):
        if isinstance(payload, str): payload = json.loads(payload)
        # dispatch to handler...

async def on_app_built(app):
    env = ServerEnv()
    if env.longtasks:
        schedule_once(0.1, env.longtasks.run)

def init():
    env = ServerEnv()
    env.longtasks = MyTasks('redis://127.0.0.1:6379', 'myqueue', worker_cnt=1, stuck_seconds=3600)
    add_startup(on_app_built)

if __name__ == '__main__':
    webapp(init)

Setup checklist:

  1. python3 -m venv ~/my-service/venv (independent venv)
  2. pip install ahserver appPublic sqlor longtasks aiohttp in the venv
  3. Write ah.py + .dspy routes following the pattern above
  4. conf/config.json with port, paths, processors
  5. Start: cd ~/my-service && source venv/bin/activate && nohup python ah.py > service.log 2>&1 &

Pitfall: GitHub is blocked on the GPU server. Two solutions:

  1. Clone repos locally with SOCKS5 proxy, then scp to server: def init(): env = ServerEnv() env.longtasks = MyTasks('redis://127.0.0.1:6379', 'myqueue', worker_cnt=1, stuck_seconds=3600) add_startup(on_app_built)

    if name == 'main': webapp(init)

    
**Setup checklist:**
1. `python3 -m venv ~/my-service/venv` (independent venv)
2. `pip install ahserver appPublic sqlor longtasks aiohttp` in the venv
3. Write ah.py + .dspy routes following the pattern above
4. `conf/config.json` with port, paths, processors
5. Start: `cd ~/my-service && source venv/bin/activate && nohup python ah.py > service.log 2>&1 &`

**Pitfall:** GitHub is blocked on the GPU server. Solutions:
1. SOCKS5 proxy tunnel via jump host:
   ```bash
   # Start tunnel (on local machine):
   ssh -N -D 1086 ymq@proxy-server
    1. Clone repo locally with proxy: 
      git -c http.proxy=socks5h://127.0.0.1:1086 \
    -c https.proxy=socks5h://127.0.0.1:1086 \
    clone https://github.com/user/repo.git
    2. Copy to server: 
      scp -r repo user@server:~/destination/

    See references/gpu-service-longtasks-pattern.md for the wan22 video generation service example (DEPRECATED — replaced by wan27). See references/wan22-deployment.md for Wan2.2 model-specific deployment issues and solutions (DEPRECATED). See references/ssh-access-map.md for SSH access architecture (which domains are passwordless, server inventory).

⚠️ SSH connection throttling on GPU server

⚠️ startswiths config: must use registerfunction key, not path

Problem: Adding a startswiths entry with {"leading": "/api/foo", "path": "/app/api/foo/"} causes KeyError: 'registerfunction' on every request to that route.

Cause: FunctionProcessor.path_call() unconditionally reads self.config_opts['registerfunction']. The path key is NOT a valid alternative — it's used by a different routing mechanism (processor-based routing via paths config).

Fix: Always use registerfunction for startswiths entries:

"startswiths": [
    {"leading": "/idfile", "registerfunction": "idfile"},
    {"leading": "/api/asr", "registerfunction": "asr"},
    {"leading": "/api/transcribe", "registerfunction": "transcribe"}
]

For new API endpoints, use the directory/index.dspy pattern instead of startswiths — the processor-based routing handles app/api/foo/index.dspy automatically via paths + indexes config. Only use startswiths with registerfunction for function-based handlers registered via RegisterFunction (like idfile, asr, etc. that are defined in Python, not .dspy files).

⚠️ idfile download endpoint requires explicit setup

Problem: Requesting /idfile?path=... returns 500 "invalid path" even though the file exists.

Cause: The idfile endpoint is registered by ahserver/filedownload.py, but it must be explicitly imported AND configured.

Fix — TWO steps required:

  1. Import in ah.py:
from ahserver import filedownload  # Registers 'idfile' and 'download' with RegisterFunction
  1. Add to conf/config.json:
"startswiths": [
    {"leading": "/idfile", "registerfunction": "idfile"},
    {"leading": "/api/...", "registerfunction": "..."}
]

How it works: filedownload.py calls rf.register('idfile', path_download) at import time. path_download() resolves the path query param via FileStorage().realPath() and returns a file_response. The startswiths config routes /idfile to FunctionProcessor which looks up the registered function.

URL encoding: Chinese characters in the path query param MUST be URL-encoded (%E9%94%99%E9%A2%91 not 错频). The server rejects unencoded Chinese in query strings.

⚠️ appPublic.worker missing schedule functions (legacy versions only)

Problem: appPublic.worker in versions ≤5.2.x (e.g. bundled with ahserver ≤1.0.8) only exports AsyncWorker — no get_event_loop, schedule_once, or schedule_interval. Code that does from appPublic.worker import schedule_once will crash with ImportError.

Fixed in appPublic ≥5.3.0. All three functions are now available. When upgrading, also upgrade apppublic: pip install --upgrade apppublic.

Workaround for environments stuck on old versions:

def get_event_loop():
    return asyncio.get_event_loop()

def schedule_once(delay, coro_func):
    async def _delayed():
        await asyncio.sleep(delay)
        await coro_func()
    asyncio.ensure_future(_delayed())

def schedule_interval(interval, coro_func):
    async def _loop():
        while True:
            await asyncio.sleep(interval)
            await coro_func()
    asyncio.ensure_future(_loop())

Applies to: longtasks module and any code using appPublic.worker scheduling. After patching, clear __pycache__ dirs.

⚠️ stream_response re-raises ClientConnectionResetError — noisy logs + unnecessary exceptions

Problem: ahserver/globalEnv.py stream_response() wraps all write errors in a generic Exception and re-raises, including ClientConnectionResetError (client disconnected mid-stream). This floods logs with stack traces every time a client times out and disconnects during SSE streaming.

Traceback pattern:

aiohttp.client_exceptions.ClientConnectionResetError: Cannot write to closing transport
→ globalEnv.py:130 raise e
→ Exception: write errore=ClientConnectionResetError('Cannot write to closing transport'), d='data: ...'

Fix (commit 387726e): Catch ClientConnectionResetError separately and break the loop:

from aiohttp.client_exceptions import ClientConnectionResetError

async for d in async_data_generator():
    try:
        await res.write(...)
    except ClientConnectionResetError:
        # Client disconnected — not a server error, stop streaming
        break
    except Exception as e:
        e = Exception(f'write error{e=}, {d=}')
        exception(f'{e}\n{format_exc()}')
        raise e

When this matters: High-concurrency load tests against streaming endpoints. Without this fix, every client timeout generates a full stack trace in server logs, making it impossible to find real errors.

⚠️ @routes decorator removed in ahserver 1.2.0+ — use app.router.add_route() instead

Problem: Code using from ahserver.webapp import webapp, routes, add_startup crashes with ImportError: cannot import name 'routes'. Code that fixes the import but keeps @routes.get(...) decorators crashes with NameError: name 'routes' is not defined.

Cause: routes (an aiohttp web.RouteTableDef) was removed from ahserver.webapp in 1.2.0. The new pattern registers routes inside the on_app_built callback.

Fix — replace @routes decorators with app.router.add_route():

# ❌ OLD (ahserver ≤1.0.x)
from ahserver.webapp import webapp, routes, add_startup

@routes.get('/api/health')
async def health(request):
    return {'status': 'ok'}

# ✅ NEW (ahserver ≥1.2.0)
from ahserver.webapp import webapp
from ahserver.configuredServer import add_startup

async def health(request):
    return {'status': 'ok'}

async def on_app_built(app):
    app.router.add_route('GET', '/api/health', health)
    # ... other routes ...

def init():
    add_startup(on_app_built)

if __name__ == '__main__':
    webapp(init)

Migration regex for bulk-fixing old services:

# Find all @routes decorators
grep -rn '@routes\.' /data/ymq/*/ah.py
# Remove decorator lines, keep function defs, register in on_app_built

⚠️ CRITICAL: Comments containing @ patterns can truncate files on production

Incident (2026-05-26): Line 134 of auth_api.py contained:

# redis = await aioredis.from_url("redis://127.0.0.1:6379")

When this comment was preceded by @web.middleware text (from the previous line's content), the deployment process truncated the file at this line, losing ~30 lines of critical code including:

  • aiohttp_session.setup(app, storage)
  • auth.setup(app, policy)
  • app.middlewares.append(self.checkAuth)

Symptom: All requests return 500. checkAuth middleware never fires — no auth logs appear. The site is completely broken.

Prevention:

  • Never put @something patterns inside comments in Python files, especially strings containing @
  • If you must reference a URL with @ in a comment, use a placeholder like redis://127.0.0.1:6379 without the @user:pass portion
  • After deploying auth_api.py changes, always verify that setupAuth() is intact: grep -c "middlewares.append" auth_api.py should return >= 1

How to verify middleware is registered:

# Check that setupAuth() contains all critical calls
grep -E "middlewares.append|setup\(app|aiohttp_session.setup" auth_api.py
# Should output 3+ lines. If fewer, file is truncated.

⚠️ Jinja2 in .ui/.tmpl files — limited filter set

Problem: Using |ternary() or other non-standard Jinja2 filters in .ui files causes jinja2.exceptions.TemplateAssertionError: No filter named 'ternary'.

Cause: ahserver's Jinja2 environment does not register custom filters like ternary. Only standard Jinja2 filters are available plus ahserver's built-in functions (entire_url(), get_user(), configValue()).

Fix: Use standard Jinja2 constructs:

{# BAD — ternary filter doesn't exist #}
"{{entire_url('/path')|ternary(get_user().nick_name, '登录')}}"

{# GOOD — use if/else expression #}
"{{get_user().nick_name if get_user() else '登录'}}"

{# GOOD — just use static text #}
"管理"

Rule: In .ui JSON templates, keep Jinja2 expressions simple. Complex conditionals should be handled client-side in JavaScript.

⚠️ Do not modify sage/conf/config.json during feature development

conf/config.json is the production configuration for the Sage platform. Feature development sessions must never modify it. If a task requires config changes (DB password, session settings, etc.), the user will handle production config changes manually.

Verifying auth_api.py after any edit

After modifying auth_api.py, always verify:

# Line count should be ~193 (not fewer)
wc -l ahserver/auth_api.py
# Must have middleware registration
grep "middlewares.append" ahserver/auth_api.py
# Must have complete setupAuth method
python3 -c "import ast; ast.parse(open('ahserver/auth_api.py').read())"

Common Issues and Solutions

Python 3.12+ Compatibility

As mentioned earlier, modify aioredis files for Python 3.12+ compatibility.

Password Encryption

Database passwords in conf/config.json are AES-ECB base64 encoded using the config's password_key:

from appPublic.aes import aes_encode_b64
key = getConfig().password_key  # from config.json
encoded = aes_encode_b64(key, 'plaintext_password')

The sqlor/sor.py class calls self.unpassword() in __init__ which decodes the password via aes_decode_b64(key, password). The password must be valid AES-ECB base64 — plain text or RC4-encoded passwords will cause ValueError: The length of the provided data is not a multiple of the block length.

For quick testing, encode the password directly:

python3 -c "from appPublic.aes import aes_encode_b64; print(aes_encode_b64('YOUR_PASSWORD_KEY', 'test'))"

Also available: the legacy RC4 encoding via appPublic.rc4:

python -m ahserver.dbpassword /path/to/your/app password123

Debugging

  • Check logs in console output
  • Use format_exc() for detailed error information
  • Access request context via request._run_ns

Performance Optimization

  • Uses uvloop and httptools for high performance
  • Connection pooling for databases
  • Async file operations with aiofiles
  • Efficient JSON handling with ujson

Benchmark methodology

To compare ahserver vs pure aiohttp performance, create a test that measures both with identical static files:

# Pure aiohttp baseline
app = web.Application()
app.router.add_static("/static/", TEST_DIR)

# ahserver full chain (session + auth + ProcessorResource)
app = AHApp(client_max_size=1000000000)
res = ProcessorResource("/", TEST_DIR, processors={})
app.router.register_resource(res)
auth = AuthAPI()
await auth.setupAuth(app)

Key finding from May 2026 benchmark: On local SSD with Redis session storage and clientinfo log level:

  • Pure aiohttp: avg 1.23ms for 48KB static file
  • ahserver (before fixes): avg 1.13ms — actually faster than aiohttp baseline
  • ahserver (after all 3 fixes): avg 1.10ms — only 0.03ms improvement

This means the 17-second delay in production is NOT caused by ahserver code overhead. The bottleneck is likely:

  • Network/proxy layer (Nginx reverse proxy, TLS handshake, keep-alive issues)
  • DNS resolution
  • Connection pool exhaustion
  • Something in the production deployment not present in local testing

Recommended production diagnostics:

  1. Use browser Network panel to check TTFB (Time To First Byte)
  2. Check if delay is in connection establishment or data transfer
  3. Add timing middleware to measure each layer: 
    @web.middleware
async def timing_middleware(request, handler):
    t0 = time.time()
    resp = await handler(request)
    t1 = time.time()
    print(f'[{request.path}] total={t1-t0:.3f}s')
    return resp
  4. Compare response time with and without Nginx in the path
  5. Check auth_api.py timecost log: if AUTH_MS is small but total is slow, bottleneck is NOT in ahserver

Multi-Service Deployment

See references/multi-service-media-platform.md for the architecture pattern used when deploying multiple ahserver instances behind nginx — covers IP filtering via geo block, async (longtasks) vs sync (lock) service split, port allocation conventions, and inter-service localhost calls.

Pipeline Orchestration Principle

Services are capability centers, not agents. When building pipelines that span multiple services (e.g., KTV song production: lyrics → music → video → subtitle → merge), the orchestration logic belongs in the calling agent/process, not embedded in one of the services.

Each service should:

  • Expose atomic capabilities via REST APIs
  • Accept file uploads or paths as input
  • Return results (file paths, URLs, status)
  • Not know about upstream/downstream steps

The agent/orchestrator should:

  • Call services in sequence
  • Handle state transitions and error recovery
  • Be the file router — download from one service, upload to the next. Services must NOT directly copy files between each other, even when co-located on the same machine. This ensures the pipeline works across distributed deployments.
  • Implement retry/threshold logic

Why: This keeps services focused, testable, and reusable. The agent has full context and can adapt the pipeline based on intermediate results.

See references/nginx-proxy-ssl-template.md for nginx proxy config template (SSL on non-standard port, geo-block IP filtering, X-Forwarded headers) and Let's Encrypt DNS-01 certificate workflow for environments where ports 80/443 are unavailable.

See references/api-routing-patterns.md for the complete guide to ahserver URL routing: directory/index.dspy pattern, nginx proxy_pass trailing slash semantics, startswiths vs processor lookup, and 1.0.x → 1.2.0 migration cheat sheet.

Session & Cookie Timeout

See references/session-timeout-architecture.md for the two-layer session system (aiohttp_session + aiohttp_auth ticket), timeout/renewal semantics, and how to configure sliding-window ticket renewal.

Hot Reload

ahserver supports hot-reloading of cached resources without restart. Each worker process independently watches file mtimes via stat(), so it works safely with reuse_port=True multi-process deployment — no Redis pub/sub, signals, or cross-process coordination needed.

What auto-reloads without config

  • .dspy files — read from disk on every request (no cache)
  • .md files — read from disk on every request (no cache)
  • .tmpl / .ui files — Jinja2 auto_reload checks mtime natively

What needs hot_reload config

  • config.json — JsonConfig is a singleton; cleared on change, next getConfig() reloads
  • i18n/*/msg.txt — MiniI18N is a singleton + ServerEnv.myi18n cache; both cleared on change

Configuration (conf/config.json)

"hot_reload": true                              // enable with default 2s interval
"hot_reload": {"enabled": true, "interval": 5}  // custom interval
// omit or false                                // disabled (default)

Architecture

Module: ahserver/hotreload.py

  • FileWatcher — tracks mtime per path, returns changed paths
  • HotReloader — on change: sets JsonConfig.instance = None and MiniI18N.instance = None (SingletonDecorator pattern)
  • hot_reload_task(app, reloader) — asyncio background task registered via app.on_startup
  • Throttled by configurable interval (default 2s) to avoid excessive stat() calls

Version History

1.3.0

  • Hot-reload module, file watching, /hot_reload endpoint, invalidate_all_caches()

1.2.0 (BREAKING CHANGES)

  • RegisterCoroutine REMOVED → use add_startup(callback) instead
  • .dspy auto-extension REMOVED → /api/status no longer resolves to /api/status.dspy; needs nginx rewrite or explicit startswiths in config
  • Added uvloop and httptools for performance improvement
  • appPublic.worker now exports schedule_once, schedule_interval, get_event_loop (was only AsyncWorker in ≤5.2.x)

1.0.8

  • Added server_error(errcode) global function for HTTP error handling
  • Added request._run_ns for accessing global environment variables

API Testing

See references/api-testing-patterns.md for testing ahserver APIs with Python urllib instead of curl — avoids shell escaping issues with Bearer tokens and JSON payloads.

vLLM Integration

See references/vllm-integration-pattern.md for integrating vLLM's AsyncLLMEngine with ahserver for high-throughput LLM inference. Covers:

  • Complete server implementation using AsyncLLMEngine + stream_response
  • Multi-GPU deployment pattern (8 GPUs × 2 instances)
  • Critical parameters: --max-model-len, --max-num-seqs, --gpu-memory-utilization
  • Common pitfalls: heredoc variable expansion, ClientConnectionResetError handling, dead DB connections after OOM
  • Migration guide from transformers-based inference to vLLM

Hot Reload

ahserver has a built-in hot-reload system (ahserver/hotreload.py) that watches file mtimes and clears cached resources. Multi-process safe — each worker process runs its own FileWatcher, no cross-process coordination needed.

Configuration

"hot_reload": true
// or with custom interval:
"hot_reload": {"enabled": true, "interval": 5}

What Gets Hot-Reloaded

Automatic (file mtime detection):

File Cache Cleared
conf/config.json JsonConfig singleton + all module caches
i18n/*/msg.txt MiniI18N singleton + ServerEnv.myi18n
.tmpl/.ui files Jinja2 auto_reload (built-in)
.dspy/.md files No cache — read from disk every request

Manual (HTTP endpoint, only when hot_reload enabled):

GET /__hot_reload__ — triggers invalidate_all_caches() which clears:

Module Cache Object Clear Method
rbac UserPermissions.ur_caches + rp_caches LRU.clear() + invalidate_rp_cache()
pricing PricingProgram.pricing_data dict.clear()
uapi UAPIData.apidata + org_users dict.clear()
llmage _uapi_cache + _uapiio_cache invalidate_uapi_cache()

Each module cleared independently with try/except — one module's import failure won't block others.

Multi-Process Caveat

With reuse_port=True, GET /__hot_reload__ only clears the single worker handling the request. Hit multiple times or rely on file-based hot-reload which works across all workers independently.

Module Cache Architecture

Module TTL Auto-invalidation
rbac.ur_caches 5min DB events
rbac.rp_caches 10min DB events
pricing.pricing_data none DB events
uapi.UAPIData none none — use /hot_reload
llmage._uapi_cache 5min invalidate_uapi_cache()

⚠️ uapi.UAPIData has no TTL and no auto-invalidation.

Version History

1.3.0

  • Hot-reload module, file watching, /hot_reload endpoint, invalidate_all_caches()

1.2.0

  • Added uvloop and httptools for performance improvement

1.0.8

  • Added server_error(errcode) global function for HTTP error handling
  • Added request._run_ns for accessing global environment variables