tao.chen/model-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add DEVELOP.md

Browse Source
This commit is contained in:
tao.chen
2026-07-31 13:34:55 +08:00
parent 5cbbe8213a
commit 2e066db04a
1 changed files with 378 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
DEVELOP.md
+378
View File
@@ -0,0 +1,378 @@
# DEVELOP.md — Developer Guide
This guide is for engineers working on the model platform codebase. For
high-level design see `ARCHITECTURE.md`; for the current state of in-flight
refactors see `HANDOVER.md`.
## Code layout
```
common/ Pure-Python shared library
config.py Settings (pydantic-settings, lru_cache singleton)
db/ SQLAlchemy 2.0 async engine, session_scope, Base
db/models/ 26 tables in 9 domain files (zero FK, zero relationship)
scheduler/ build_sqlalchemy_jobstore (delayed import)
storage/ RustFSObjectStore + StorageClient + Pydantic schemas
eventing.py add_outbox_event / utcnow / event_time
service_app.py /health/ready TCP probe, /api/v1/health
schemas.py StrictModel base
utils.py get_free_port, start_process
backend/ Public FastAPI service + internal /internal/v1/* sub-app
main.py lifespan + route registration
jupyter.py /api/v1/auth/jupyter — the ONLY auth entry
scripts.py CRUD for scripts/notebooks (workspace_fs=rustfs)
schedules.py DAG CRUD: schedules, nodes, edges
schedule_runs.py Trigger / list / get runs
schedule_schemas.py Pydantic request/response models
admin.py Admin endpoints
resources.py Misc data resources
storage_api.py /internal/v1/* (sub-app merged into main)
storage_client.py HTTP client for the storage sub-app
schedule_client.py Placeholder module (was the HTTP-push executor client)
runtime_client.py Self-contained httpx wrapper for the runtime
jupyter.py auth_request handler
dependencies.py request_context, database_session
schedule/ Schedule Executor (DAG worker)
context.py Constants + naive_utc
scheduler.py CronScheduler (APScheduler + 5s sync loop)
orchestrator.py DispatchOrchestrator (Outbox poll + DAG advance)
worker.py NodeExecutor (notebook / python execution)
service.py SchedulerService facade (composes the three)
main.py Lifespan + FastAPI app
storage_client.py SchedulerStorageClient (subclass of common StorageClient)
execution.py execute_artifact (notebook + python paths)
notebook_runner.py Subprocess entry point (nbclient)
runtime/ Jupyter Runtime
main.py FastAPI entry: jupyter action endpoints
process.py Per-workspace subprocess pool + asyncio locks
mount.py rclone FUSE mount lifecycle
frontend/ React Router SPA (vite build → nginx)
app/ features/ routes/ services/ components/
migrations/ Alembic schema versions
docker-compose.yml 4 services
default.conf Nginx template
scripts/nginx-entrypoint.sh
.env.example
```
## Configuration system
All env vars go through one place: `common/src/common/config.py`.
```python
from common.config import settings
settings.database_url # str
settings.rustfs_endpoint # str (full URL, e.g. "http://rustfs:9000")
settings.rustfs_access_key # str
settings.rustfs_secret_key # str
settings.rustfs_workspace_bucket
settings.rustfs_version_bucket
settings.rustfs_run_log_bucket
settings.jwt_secret # HS256 secret for the auth_request handler
settings.workspace_root # schedule subprocess cwd; backend ignores
settings.workspaces_root # runtime rclone FUSE mount point
settings.remote_bucket # rclone remote spec (e.g. "rustfs:workspaces")
settings.backend_api_url # schedule → backend HTTP base
settings.runtime_api_url # backend → runtime HTTP base
settings.public_base_url # runtime public base URL
settings.service_name # surfaced in /health
settings.readiness_targets # CSV host:port list for /health/ready
```
`Settings` reads from process env first, then from a `.env` file at CWD
if present. `pydantic-settings` auto-loads. `case_sensitive=False` so
`DATABASE_URL` / `database_url` both work.
### Adding a new env var
1. Add the field to `Settings`:
```python
new_var: str = Field(default="x", description="...")
```
2. Add the line to `.env.example` with a comment.
3. Use `settings.new_var` at the call site.
Do **not** call `os.environ["NEW_VAR"]` or `os.getenv("NEW_VAR")` in
application code. The grep below should return zero hits:
```bash
grep -rnE 'os\.(environ\[?["\x27][A-Z_]+|getenv\(["\x27][A-Z_]+)' --include="*.py" \
backend/src/ schedule/src/ runtime/src/ common/src/
```
## Conventions
### Database / SQLAlchemy
- **No foreign keys, no `relationship`** — every join is explicit.
- Every table has `is_deleted TINYINT(1) NOT NULL DEFAULT 0` and
`deleted_at DATETIME(3) NULL`; queries must filter `is_deleted == 0`
(or `deleted_at.is_(None)`) to avoid logical-deleted rows.
- Domain files under `common/src/common/db/models/` are split by
bounded context: `audit` / `events` / `experiments` / `identity` /
`runtime` / `schedules` / `scripts` / `storage` / `workspaces`.
- All models are `class X(Base)` SQLAlchemy 2.0 declarative-mapped.
- The full schema is in `migrations/versions/`. Apply with:
```bash
uv run --frozen --package backend alembic upgrade head
```
### Storage
- All object bytes go to **RustFS** via boto3.
- Use `StorageClient` (HTTP) or `RustFSObjectStore` (direct) — never
the local filesystem.
- `bucket_name` is one of: `RUSTFS_WORKSPACE_BUCKET` (default
`workspaces`), `RUSTFS_VERSION_BUCKET` (default `versions`,
reserved), `RUSTFS_RUN_LOG_BUCKET` (default `run-logs`, reserved).
### Auth
- Browser → `/jupyter/{workspace_id}/...` → Nginx `auth_request` →
`GET /api/v1/auth/jupyter` (Backend).
- The handler:
1. Parses cookie / Bearer JWT (HS256 + `settings.jwt_secret`).
2. Verifies `WorkspaceMembers` for the workspace.
3. Verifies `Scripts.is_locked` for the requested notebook path
(owner / unlocked → allow; otherwise 403).
4. Calls `RuntimeClient.get_workspace` / `start_workspace`.
5. Returns `x-upstream-addr` + `x-jupyter-internal-token` response
headers. **Browser never holds the runtime token.**
- Nginx captures the headers via `auth_request_set` and proxies to the
upstream sub-process with `Authorization: token $jupyter_token`.
### Permission gates
- For write operations on a script/notebook, call
`require_script_modify_access(script, user_id=..., is_admin=...)`
from `backend/scripts.py`. It enforces:
- admin or owner → allow
- non-owner, `is_locked == 0` → allow
- non-owner, `is_locked == 1` → 403
- Read endpoints (`list_scripts`, `get_script`) intentionally do **not**
check `is_locked` — workspace members can see the script list.
### Outbox events
- The platform's only async-messaging fabric is the MySQL
`OutboxEvents` table. Producers (Backend) write rows in the same
transaction as the business state. Consumers (Schedule Executor)
poll every 250 ms and update `event_status` to `published` or
`failed` (with retry).
- Use `common.eventing.add_outbox_event` to write.
- `event_type` values currently in use:
- `schedule.run.requested` — produced by `schedule_runs.py` / cron post-back
- `job.node.execute` — produced by orchestrator when a node is ready
- `job.node.finished` — produced by worker after node execution
- `consumer_inbox` provides exactly-once delivery per
`(consumer_name, event_id)` (with `process_status: processing →
succeeded` lifecycle).
### Async / sync signatures
- `SchedulerService.start()` and `SchedulerService.close()` are
`async def` (so the FastAPI lifespan can `await` them).
- `CronScheduler.start()`, `DispatchOrchestrator.start()` are
**`def` (sync)** — they only `create_task(...)` and return. Don't
`await` them.
- `cron.start()`, `orchestrator.start()`, `worker.handle_node_execute`
are wired together in `SchedulerService.__init__`; their lifetime
is owned by the facade.
## Local development
### One-time setup
```bash
# Python workspace (monorepo via uv workspaces)
uv sync --all-packages
# Frontend deps
cd frontend && pnpm install && cd ..
```
### Per-service dev
```bash
# Backend (terminal 1)
export DATABASE_URL="mysql+asyncmy://model_platform:model_platform@127.0.0.1:3306/model_platform?charset=utf8mb4"
export RUSTFS_ACCESS_KEY=modelplatform
export RUSTFS_SECRET_KEY=modelplatformsecret
export RUSTFS_ENDPOINT=http://127.0.0.1:9000
uv run --frozen --package backend uvicorn backend.main:app --host 0.0.0.0 --port 8000 --reload
# Schedule Executor (terminal 2)
uv run --frozen --package schedule uvicorn schedule.main:app --host 0.0.0.0 --port 8001 --reload
# Runtime (terminal 3 — needs SYS_ADMIN, FUSE, devmode)
uv run --frozen --package runtime python -m runtime.main
```
### Frontend dev
```bash
cd frontend
pnpm dev # http://localhost:5173, proxies /api to backend
pnpm typecheck
pnpm build
```
### Static checks
```bash
# Python compile
uv run --frozen --package backend python -m compileall -q backend/src common/src
uv run --frozen --package schedule python -m compileall -q schedule/src
uv run --frozen --package runtime python -m compileall -q runtime/src
# Type check (frontend)
cd frontend && pnpm typecheck && cd ..
# Docker compose config
docker compose config --quiet
```
### Smoke test
```bash
# Run the full ORM import + Settings smoke test
PYTHONPATH="backend/src:common/src" uv run --frozen --package backend python -c "
from backend.main import app
from common.config import settings
print('backend:', len(app.routes), 'routes')
print('settings ok:', settings.rustfs_endpoint)
"
```
## Common tasks
### Add a new DAG endpoint
1. Add the route handler in `backend/schedules.py` (DAG template) or
`backend/schedule_runs.py` (run lifecycle).
2. Validate request via `backend/schedule_schemas.py`.
3. For mutations on nodes/edges/versions: route through
`create_script_record` / `get_script_row` and apply
`require_script_modify_access` if it touches a script.
4. If it produces an outbox event, use
`add_outbox_event(session, event_type="...", producer="...", ...)`.
### Add a new env var
See "Adding a new env var" above.
### Add a new MySQL table
1. Add a model class in `common/src/common/db/models/<domain>.py`.
Include `is_deleted TINYINT(1) NOT NULL DEFAULT 0` and
`deleted_at DATETIME(3) NULL`.
2. Export it from `common/src/common/db/models/__init__.py`.
3. Generate the migration:
```bash
uv run --frozen --package backend alembic revision --autogenerate -m "add <feature>"
```
4. Review the generated `migrations/versions/*.py` — Alembic may
miss comments / server defaults. Manually fix the migration.
5. Apply locally:
```bash
uv run --frozen --package backend alembic upgrade head
```
### Wire a new RustFS bucket
1. Add the env var to `Settings`:
```python
rustfs_<feature>_bucket: str = Field(default="<feature>", description="...")
```
2. Add to `.env.example`.
3. Extend the `Literal` in `ServerObjectRequest.usage_type` to include
the new `usage_type` value, if applicable.
4. In the consumer of the bucket, branch on `usage_type` (or the
`bucket_name` argument) and pick the right bucket via
`settings.rustfs_<feature>_bucket`.
### Add a new schedule node type
`schedule/execution.py` dispatches on `script_type` in
`execute_artifact`. Add a new branch + a new `_<type>` function.
`worker.py` does not need to change — the dispatch happens inside
`execute_artifact`.
## Tests
There is **no formal test suite yet** (see HANDOVER §Pending Tasks
P1). A reasonable first test surface:
- `require_script_modify_access` (admin / owner / non-owner-unlock /
non-owner-lock): pure-function unit test, no DB.
- `validate_dag` (cycle detection + orphan detection) in
`backend/schedules.py`.
- `execute_artifact` end-to-end with mocked `content_hash` and a
real `tempfile.TemporaryDirectory`.
Test convention: pytest with `pytest-asyncio` for `async def`
handlers. Use SQLite in-memory (or a MySQL test container) for DB
integration. Use moto for S3.
## Troubleshooting
### "the greenlet library is required"
SQLAlchemy 2.0 needs `greenlet` for `engine.dispose()` in async
contexts. Add `greenlet>=3.0.0` to `common/pyproject.toml` and
`uv sync --all-packages`. (Already present in this repo.)
### "Can't connect to MySQL server"
Either MySQL isn't running, or the network namespace doesn't allow
`mysql:3306` resolution. Inside the Docker network, services reach
each other by service name (`mysql`, `backend`, `runtime`,
`schedule`, `rustfs`).
### Jupyter routing 401s
Inspect `docker compose logs backend` — `jupyter.py:check_notebook_is_locked`
or `load_active_membership` will return an explicit reason. Then
check the JWT (use `JWT_SECRET` from `.env`).
### Schedule run never advances
`schedule_runs.run_status` is stuck at `queued`. Two likely causes:
- `outbox_events` is empty (Backend's `add_outbox_event` failed —
check `add_outbox_event` in `schedule_runs.py`).
- The orchestrator's polling loop is dead. Check
`docker compose logs schedule` and look for "database event loop
failed" exceptions.
### "AttributeError: 'SchedulerService' object has no attribute 'worker'"
`worker` must be constructed before `orchestrator` in
`SchedulerService.__init__`, because orchestrator's dispatch table
captures `self.worker.handle_node_execute` at construction time.
See `service.py` — the order is load-bearing.
## Style
- Type hints everywhere (this repo uses `from __future__ import
annotations`).
- 4-space indent, double quotes, no trailing whitespace.
- Comments are technical (explain *why*, not *what*).
- Module docstrings document non-obvious invariants. Don't add
docstrings to functions whose behavior is self-evident from the
name.
- 4 levels of indentation = "this function is doing too much; split
it". (Project convention; see e.g. `execute_artifact`.)
## See also
- `ARCHITECTURE.md` — design diagrams
- `HANDOVER.md` — current refactor state and pending work
- `CLAUDE.md` — agent-facing conventions for the repo
- `models / __init__.py` — exhaustive list of all 26 tables
- `common/config.py` — all env vars in one place