diff --git a/.env.example b/.env.example index 6e9c52e..4ea3d39 100644 --- a/.env.example +++ b/.env.example @@ -31,9 +31,15 @@ SMTP_TLS=True SMTP_SSL=False SMTP_PORT=587 -# Postgres -POSTGRES_SERVER=localhost -POSTGRES_PORT=5432 +# Where this installation keeps everything: the database, flows, secrets, +# artifacts and the user venv. The compose stack sets it to the /data volume. +DATA_DIR=flow-data +# Any SQLAlchemy URL. Empty means SQLite in the data directory, which is what +# the engine is designed around — one process owns this database. +DATABASE_URL= + +# The Postgres the optional Umami analytics profile runs on. Nothing else uses +# it; `make umami` provisions a least-privilege role inside it. POSTGRES_DB=app POSTGRES_USER=postgres POSTGRES_PASSWORD=changethis diff --git a/.gitea/workflows/test-backend.yml b/.gitea/workflows/test-backend.yml index 30bad56..4c17551 100644 --- a/.gitea/workflows/test-backend.yml +++ b/.gitea/workflows/test-backend.yml @@ -39,30 +39,16 @@ jobs: pyproject.toml backend/pyproject.toml uv.lock - - name: Write .env - # Not committed, and both compose and the backend settings need it. - # The suite runs on the runner and reaches the container over - # POSTGRES_SERVER=localhost from there, on the port resolved below. - run: cp .env.example .env - - run: docker compose down -v --remove-orphans - # Postgres is the only service the suite needs: the mail paths are - # patched in tests/, and the flow engine falls back to in-memory state - # while REDIS_HOST is empty. - - run: docker compose up -d --wait db - - name: Resolve the db port - # Docker assigns it (docker/compose.ci.yml), so concurrent runs get one - # each. An env var outranks the .env file in the backend settings. - run: | - echo "POSTGRES_PORT=$(docker compose port db 5432 | tail -n1 | cut -d: -f2)" >> "$GITHUB_ENV" - run: uv sync - - name: Migrate DB and seed the superuser - run: uv run bash scripts/prestart.sh - working-directory: backend - # scripts/tests-start.sh waits for the DB, then runs pytest under - # coverage and prints the report (see backend/scripts/test.sh). + # Nothing to start: the suite names its own SQLite file in a temp + # directory (backend/tests/__init__.py), the mail paths are patched, and + # the flow engine falls back to in-memory state without a Redis host. - name: Run tests with coverage - run: uv run bash scripts/tests-start.sh "Coverage for ${{ github.sha }}" + run: uv run bash scripts/test.sh "Coverage for ${{ github.sha }}" working-directory: backend - - name: Tear down - if: always() - run: docker compose down -v --remove-orphans + - name: Check the wheels build + # Cheap, and it catches a data file that stopped being packaged — + # alembic's revisions and the email templates ride along by directory. + run: | + uv build --all-packages --out-dir dist + uv run --no-project --with dist/fluksio_worker-*.whl fluksio-worker --help diff --git a/.gitea/workflows/test-compose.yml b/.gitea/workflows/test-compose.yml index 4bb3761..d7758d1 100644 --- a/.gitea/workflows/test-compose.yml +++ b/.gitea/workflows/test-compose.yml @@ -47,8 +47,9 @@ jobs: # breaks silently. - run: docker compose build backend frontend - run: docker compose down -v --remove-orphans - # db and prestart come along as depends_on; --wait blocks on the backend - # healthcheck in docker/compose.yml. + # Redis comes along as a depends_on; --wait blocks on the backend + # healthcheck in docker/compose.yml, which is also what proves the + # schema was created before anything served a request. - run: docker compose up -d --wait backend frontend # Nothing is published on the host (compose.ci.yml), so the checks run # from inside the backend container against the project network. diff --git a/Makefile b/Makefile index ff8a120..6cd4738 100644 --- a/Makefile +++ b/Makefile @@ -26,15 +26,15 @@ help: ## Show available targets # ── Development (Docker) ────────────────────────────────────────── -dev-utils: ## Start only the utility containers (db, adminer, proxy, mailcatcher, prestart) - $(COMPOSE_DEV) up --build db adminer proxy mailcatcher prestart +dev-utils: ## Start only the utility containers (proxy, mailcatcher) + $(COMPOSE_DEV) up --build proxy mailcatcher dev: ## Start the full dev stack (includes a local Traefik proxy) $(COMPOSE_DEV) up --build dev-local: ## Start the integrated local stack (called by the root `make dev`) DOMAIN=$${DOMAIN:-localhost} ENVIRONMENT=$${ENVIRONMENT:-local} \ - $(COMPOSE_LOCAL) up --build -d proxy db adminer prestart backend frontend mailcatcher + $(COMPOSE_LOCAL) up --build -d proxy backend frontend mailcatcher up: ## Start the production stack $(COMPOSE_PROD_RUN) up --build -d @@ -121,13 +121,10 @@ seed-hosted-demo: ## Seed the hosted demo panel (operators only; API_URL select test: test-backend test-frontend ## Run all tests (backend + frontend) -# The integrated stack publishes no host port for Postgres (docker/compose.local.yml -# drops them all), so the suite reaches the container on the compose network -# instead. Empty when the stack is down — then .env's localhost:5432 stands. -DB_HOST = $(shell docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}} {{end}}' fluksio-db 2>/dev/null | awk '{print $$1}') - test-backend: ## Run backend tests (pytest + coverage) - cd backend && $(if $(DB_HOST),POSTGRES_SERVER=$(DB_HOST)) uv run bash scripts/tests-start.sh + # Its own SQLite file in a temp directory (tests/__init__.py), so this needs + # nothing running and touches no development data. + cd backend && uv run bash scripts/test.sh # The hostname the stack is served under. A checkout configured for a # deployment carries the deployment's domain, and *.fluksio.com resolves to the diff --git a/README.md b/README.md index ce4d6ea..3672ee2 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,7 @@ repo holds the FastAPI backend, the flow engine, and the dashboard SPA. It is se ## Layout ```text -backend/ FastAPI + SQLModel + Alembic + Postgres +backend/ FastAPI + SQLModel + Alembic + SQLite fluksio/flow/ the flow engine (nodes, pipeline, state backends, controller) frontend/ React 19 + TanStack Router + Tailwind 4 + shadcn/ui docker/ compose.yml → compose.dev.yml → compose.local.yml (+ compose.traefik.yml) @@ -23,7 +23,7 @@ Normally driven from the workspace root (`make init` once, then `make dev`). Sta ```sh make install # uv sync + bun install -make dev-utils # db, adminer, proxy, mailcatcher, prestart only +make dev-utils # proxy and mailcatcher only make dev-backend # FastAPI on :8000, hot reload make dev-frontend # Vite on :5173 ``` diff --git a/backend/fluksio/alembic/env.py b/backend/fluksio/alembic/env.py index 46e7e5a..d28a18b 100755 --- a/backend/fluksio/alembic/env.py +++ b/backend/fluksio/alembic/env.py @@ -1,4 +1,3 @@ -import os from logging.config import fileConfig from alembic import context @@ -10,7 +9,9 @@ config = context.config # Interpret the config file for Python logging. # This line sets up loggers basically. -fileConfig(config.config_file_name) +# Driven from `fluksio.core.db` there is no ini file to read logging from. +if config.config_file_name is not None: + fileConfig(config.config_file_name) # add your model's MetaData object here # for 'autogenerate' support @@ -20,9 +21,23 @@ fileConfig(config.config_file_name) from fluksio.models import SQLModel # noqa from fluksio.core.config import settings # noqa +from fluksio.core.types import UTCDateTime # noqa target_metadata = SQLModel.metadata + +def render_item(type_, obj, autogen_context): + """Write the decorator out as the type it wraps. + + A revision naming `UTCDateTime` would have to import from the app, which + ties a migration to a version of it. What it stores is a timezone-aware + timestamp, so that is what the revision says. + """ + if type_ == "type" and isinstance(obj, UTCDateTime): + autogen_context.imports.add("import sqlalchemy as sa") + return "sa.DateTime(timezone=True)" + return False + # other values from the config, defined by the needs of env.py, # can be acquired: # my_important_option = config.get_main_option("my_important_option") @@ -61,6 +76,13 @@ def run_migrations_online(): and associate a connection with the context. """ + # `fluksio.core.db.migrate` hands us the connection it is already inside; + # the CLI and `alembic` on the command line make their own. + connection = config.attributes.get("connection") + if connection is not None: + _run(connection) + return + configuration = config.get_section(config.config_ini_section) configuration["sqlalchemy.url"] = get_url() connectable = engine_from_config( @@ -70,12 +92,20 @@ def run_migrations_online(): ) with connectable.connect() as connection: - context.configure( - connection=connection, target_metadata=target_metadata, compare_type=True - ) + _run(connection) - with context.begin_transaction(): - context.run_migrations() + +def _run(connection): + context.configure( + connection=connection, + target_metadata=target_metadata, + compare_type=True, + render_item=render_item, + # SQLite cannot ALTER much: batch mode rebuilds the table instead. + render_as_batch=connection.dialect.name == "sqlite", + ) + with context.begin_transaction(): + context.run_migrations() if context.is_offline_mode(): diff --git a/backend/fluksio/alembic/versions/087c44e16304_add_observability_rollups_events_and_.py b/backend/fluksio/alembic/versions/087c44e16304_add_observability_rollups_events_and_.py deleted file mode 100644 index a35f29a..0000000 --- a/backend/fluksio/alembic/versions/087c44e16304_add_observability_rollups_events_and_.py +++ /dev/null @@ -1,77 +0,0 @@ -"""Add observability rollups, events and runs - -Revision ID: 087c44e16304 -Revises: 59e2606ce144 -Create Date: 2026-08-16 19:55:10.772691 - -""" -from alembic import op -import sqlalchemy as sa -import sqlmodel.sql.sqltypes - - -# revision identifiers, used by Alembic. -revision = '087c44e16304' -down_revision = '59e2606ce144' -branch_labels = None -depends_on = None - - -def upgrade(): - # ### commands auto generated by Alembic - please adjust! ### - op.create_table('engine_event', - sa.Column('id', sa.Integer(), nullable=False), - sa.Column('ts', sa.DateTime(timezone=True), nullable=False), - sa.Column('type', sqlmodel.sql.sqltypes.AutoString(length=32), nullable=False), - sa.Column('flow', sqlmodel.sql.sqltypes.AutoString(), nullable=False), - sa.Column('node', sqlmodel.sql.sqltypes.AutoString(), nullable=False), - sa.Column('detail', sqlmodel.sql.sqltypes.AutoString(), nullable=False), - sa.Column('actor', sqlmodel.sql.sqltypes.AutoString(), nullable=False), - sa.PrimaryKeyConstraint('id') - ) - op.create_index(op.f('ix_engine_event_ts'), 'engine_event', ['ts'], unique=False) - op.create_index(op.f('ix_engine_event_type'), 'engine_event', ['type'], unique=False) - op.create_table('flow_run', - sa.Column('id', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False), - sa.Column('flow', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False), - sa.Column('source', sqlmodel.sql.sqltypes.AutoString(), nullable=False), - sa.Column('started_at', sa.DateTime(timezone=True), nullable=False), - sa.Column('finished_at', sa.DateTime(timezone=True), nullable=True), - sa.Column('status', sqlmodel.sql.sqltypes.AutoString(), nullable=False), - sa.Column('nodes', sa.Integer(), nullable=False), - sa.Column('errors', sa.Integer(), nullable=False), - sa.Column('duration_ms', sa.Float(), nullable=False), - sa.Column('deliveries', sa.Integer(), nullable=False), - sa.PrimaryKeyConstraint('id') - ) - op.create_index(op.f('ix_flow_run_flow'), 'flow_run', ['flow'], unique=False) - op.create_index(op.f('ix_flow_run_started_at'), 'flow_run', ['started_at'], unique=False) - op.create_table('metric_minute', - sa.Column('flow', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False), - sa.Column('node', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False), - sa.Column('bucket', sa.DateTime(timezone=True), nullable=False), - sa.Column('executions', sa.Integer(), nullable=False), - sa.Column('errors', sa.Integer(), nullable=False), - sa.Column('messages', sa.Integer(), nullable=False), - sa.Column('duration_sum_ms', sa.Float(), nullable=False), - sa.Column('duration_max_ms', sa.Float(), nullable=False), - sa.Column('lag_sum_ms', sa.Float(), nullable=False), - sa.Column('lag_max_ms', sa.Float(), nullable=False), - sa.Column('items', sa.Integer(), nullable=False), - sa.PrimaryKeyConstraint('flow', 'node', 'bucket') - ) - op.create_index(op.f('ix_metric_minute_bucket'), 'metric_minute', ['bucket'], unique=False) - # ### end Alembic commands ### - - -def downgrade(): - # ### commands auto generated by Alembic - please adjust! ### - op.drop_index(op.f('ix_metric_minute_bucket'), table_name='metric_minute') - op.drop_table('metric_minute') - op.drop_index(op.f('ix_flow_run_started_at'), table_name='flow_run') - op.drop_index(op.f('ix_flow_run_flow'), table_name='flow_run') - op.drop_table('flow_run') - op.drop_index(op.f('ix_engine_event_type'), table_name='engine_event') - op.drop_index(op.f('ix_engine_event_ts'), table_name='engine_event') - op.drop_table('engine_event') - # ### end Alembic commands ### diff --git a/backend/fluksio/alembic/versions/1a31ce608336_add_cascade_delete_relationships.py b/backend/fluksio/alembic/versions/1a31ce608336_add_cascade_delete_relationships.py deleted file mode 100644 index 10e47a1..0000000 --- a/backend/fluksio/alembic/versions/1a31ce608336_add_cascade_delete_relationships.py +++ /dev/null @@ -1,37 +0,0 @@ -"""Add cascade delete relationships - -Revision ID: 1a31ce608336 -Revises: d98dd8ec85a3 -Create Date: 2024-07-31 22:24:34.447891 - -""" -from alembic import op -import sqlalchemy as sa -import sqlmodel.sql.sqltypes - - -# revision identifiers, used by Alembic. -revision = '1a31ce608336' -down_revision = 'd98dd8ec85a3' -branch_labels = None -depends_on = None - - -def upgrade(): - # ### commands auto generated by Alembic - please adjust! ### - op.alter_column('item', 'owner_id', - existing_type=sa.UUID(), - nullable=False) - op.drop_constraint('item_owner_id_fkey', 'item', type_='foreignkey') - op.create_foreign_key(None, 'item', 'user', ['owner_id'], ['id'], ondelete='CASCADE') - # ### end Alembic commands ### - - -def downgrade(): - # ### commands auto generated by Alembic - please adjust! ### - op.drop_constraint(None, 'item', type_='foreignkey') - op.create_foreign_key('item_owner_id_fkey', 'item', 'user', ['owner_id'], ['id']) - op.alter_column('item', 'owner_id', - existing_type=sa.UUID(), - nullable=True) - # ### end Alembic commands ### diff --git a/backend/fluksio/alembic/versions/59e2606ce144_add_oauth_client_code_and_refresh_token.py b/backend/fluksio/alembic/versions/59e2606ce144_add_oauth_client_code_and_refresh_token.py deleted file mode 100644 index 65ee53e..0000000 --- a/backend/fluksio/alembic/versions/59e2606ce144_add_oauth_client_code_and_refresh_token.py +++ /dev/null @@ -1,70 +0,0 @@ -"""add oauth client, code and refresh token - -Revision ID: 59e2606ce144 -Revises: b7c41d2f8a30 -Create Date: 2026-08-15 23:59:41.147202 - -""" -from alembic import op -import sqlalchemy as sa -import sqlmodel.sql.sqltypes - - -# revision identifiers, used by Alembic. -revision = '59e2606ce144' -down_revision = 'b7c41d2f8a30' -branch_labels = None -depends_on = None - - -def upgrade(): - # ### commands auto generated by Alembic - please adjust! ### - op.create_table('oauth_client', - sa.Column('id', sa.Uuid(), nullable=False), - sa.Column('client_name', sqlmodel.sql.sqltypes.AutoString(length=128), nullable=False), - sa.Column('redirect_uris', sa.JSON(), nullable=True), - sa.Column('created_at', sa.DateTime(), nullable=False), - sa.PrimaryKeyConstraint('id') - ) - op.create_table('oauth_authorization_code', - sa.Column('id', sa.Uuid(), nullable=False), - sa.Column('code_hash', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False), - sa.Column('client_id', sa.Uuid(), nullable=False), - sa.Column('user_id', sa.Uuid(), nullable=False), - sa.Column('redirect_uri', sqlmodel.sql.sqltypes.AutoString(length=2048), nullable=False), - sa.Column('code_challenge', sqlmodel.sql.sqltypes.AutoString(length=128), nullable=False), - sa.Column('resource', sqlmodel.sql.sqltypes.AutoString(length=2048), nullable=True), - sa.Column('expires_at', sa.DateTime(), nullable=False), - sa.Column('used_at', sa.DateTime(), nullable=True), - sa.Column('refresh_token_id', sa.Uuid(), nullable=True), - sa.Column('created_at', sa.DateTime(), nullable=False), - sa.ForeignKeyConstraint(['client_id'], ['oauth_client.id'], ondelete='CASCADE'), - sa.ForeignKeyConstraint(['user_id'], ['user.id'], ondelete='CASCADE'), - sa.PrimaryKeyConstraint('id') - ) - op.create_index(op.f('ix_oauth_authorization_code_code_hash'), 'oauth_authorization_code', ['code_hash'], unique=True) - op.create_table('oauth_refresh_token', - sa.Column('id', sa.Uuid(), nullable=False), - sa.Column('token_hash', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False), - sa.Column('client_id', sa.Uuid(), nullable=False), - sa.Column('user_id', sa.Uuid(), nullable=False), - sa.Column('family_id', sa.Uuid(), nullable=False), - sa.Column('expires_at', sa.DateTime(), nullable=False), - sa.Column('revoked', sa.Boolean(), nullable=False), - sa.Column('created_at', sa.DateTime(), nullable=False), - sa.ForeignKeyConstraint(['client_id'], ['oauth_client.id'], ondelete='CASCADE'), - sa.ForeignKeyConstraint(['user_id'], ['user.id'], ondelete='CASCADE'), - sa.PrimaryKeyConstraint('id') - ) - op.create_index(op.f('ix_oauth_refresh_token_token_hash'), 'oauth_refresh_token', ['token_hash'], unique=True) - # ### end Alembic commands ### - - -def downgrade(): - # ### commands auto generated by Alembic - please adjust! ### - op.drop_index(op.f('ix_oauth_refresh_token_token_hash'), table_name='oauth_refresh_token') - op.drop_table('oauth_refresh_token') - op.drop_index(op.f('ix_oauth_authorization_code_code_hash'), table_name='oauth_authorization_code') - op.drop_table('oauth_authorization_code') - op.drop_table('oauth_client') - # ### end Alembic commands ### diff --git a/backend/fluksio/alembic/versions/9c0a54914c78_add_max_length_for_string_varchar_.py b/backend/fluksio/alembic/versions/9c0a54914c78_add_max_length_for_string_varchar_.py deleted file mode 100755 index 78a4177..0000000 --- a/backend/fluksio/alembic/versions/9c0a54914c78_add_max_length_for_string_varchar_.py +++ /dev/null @@ -1,69 +0,0 @@ -"""Add max length for string(varchar) fields in User and Items models - -Revision ID: 9c0a54914c78 -Revises: e2412789c190 -Create Date: 2024-06-17 14:42:44.639457 - -""" -from alembic import op -import sqlalchemy as sa -import sqlmodel.sql.sqltypes - - -# revision identifiers, used by Alembic. -revision = '9c0a54914c78' -down_revision = 'e2412789c190' -branch_labels = None -depends_on = None - - -def upgrade(): - # Adjust the length of the email field in the User table - op.alter_column('user', 'email', - existing_type=sa.String(), - type_=sa.String(length=255), - existing_nullable=False) - - # Adjust the length of the full_name field in the User table - op.alter_column('user', 'full_name', - existing_type=sa.String(), - type_=sa.String(length=255), - existing_nullable=True) - - # Adjust the length of the title field in the Item table - op.alter_column('item', 'title', - existing_type=sa.String(), - type_=sa.String(length=255), - existing_nullable=False) - - # Adjust the length of the description field in the Item table - op.alter_column('item', 'description', - existing_type=sa.String(), - type_=sa.String(length=255), - existing_nullable=True) - - -def downgrade(): - # Revert the length of the email field in the User table - op.alter_column('user', 'email', - existing_type=sa.String(length=255), - type_=sa.String(), - existing_nullable=False) - - # Revert the length of the full_name field in the User table - op.alter_column('user', 'full_name', - existing_type=sa.String(length=255), - type_=sa.String(), - existing_nullable=True) - - # Revert the length of the title field in the Item table - op.alter_column('item', 'title', - existing_type=sa.String(length=255), - type_=sa.String(), - existing_nullable=False) - - # Revert the length of the description field in the Item table - op.alter_column('item', 'description', - existing_type=sa.String(length=255), - type_=sa.String(), - existing_nullable=True) diff --git a/backend/fluksio/alembic/versions/a3f1c07b52d9_add_run_tables.py b/backend/fluksio/alembic/versions/a3f1c07b52d9_add_run_tables.py deleted file mode 100644 index c8b8992..0000000 --- a/backend/fluksio/alembic/versions/a3f1c07b52d9_add_run_tables.py +++ /dev/null @@ -1,126 +0,0 @@ -"""Add run, run_node, run_metric and run_artifact - -Runs are the batch shape of the engine: a flow taken from its inputs to its -outputs once, with parameters that identify it and a result worth keeping. -Deliberately its own tables rather than columns on flow_run — a cascade row is -rolled up and pruned on a retention window, and an experiment must not be. - -Revision ID: a3f1c07b52d9 -Revises: 087c44e16304 -Create Date: 2026-08-18 09:12:44.108312 - -""" -import sqlalchemy as sa -import sqlmodel.sql.sqltypes -from alembic import op - -# revision identifiers, used by Alembic. -revision = 'a3f1c07b52d9' -down_revision = '087c44e16304' -branch_labels = None -depends_on = None - - -def upgrade(): - op.create_table( - 'run', - sa.Column('id', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False), - sa.Column('flow', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False), - sa.Column('flow_version', sa.Integer(), nullable=False), - sa.Column('commit', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False), - sa.Column('params', sa.JSON(), nullable=True), - sa.Column( - 'params_digest', - sqlmodel.sql.sqltypes.AutoString(length=64), - nullable=False, - ), - sa.Column('seed', sa.Integer(), nullable=True), - sa.Column('group_id', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=True), - sa.Column('parent_id', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=True), - sa.Column('cause', sqlmodel.sql.sqltypes.AutoString(length=32), nullable=False), - sa.Column('status', sqlmodel.sql.sqltypes.AutoString(length=16), nullable=False), - sa.Column( - 'status_reason', - sqlmodel.sql.sqltypes.AutoString(length=1024), - nullable=False, - ), - sa.Column('labels', sa.JSON(), nullable=True), - sa.Column('created_at', sa.DateTime(timezone=True), nullable=False), - sa.Column('started_at', sa.DateTime(timezone=True), nullable=True), - sa.Column('finished_at', sa.DateTime(timezone=True), nullable=True), - sa.Column('duration_ms', sa.Float(), nullable=False), - sa.Column('result', sa.JSON(), nullable=True), - sa.Column('engine', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False), - sa.Column('lease_at', sa.DateTime(timezone=True), nullable=True), - sa.Column('actor', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False), - sa.PrimaryKeyConstraint('id'), - ) - op.create_index(op.f('ix_run_flow'), 'run', ['flow'], unique=False) - op.create_index(op.f('ix_run_status'), 'run', ['status'], unique=False) - op.create_index(op.f('ix_run_group_id'), 'run', ['group_id'], unique=False) - op.create_index(op.f('ix_run_created_at'), 'run', ['created_at'], unique=False) - op.create_index( - op.f('ix_run_params_digest'), 'run', ['params_digest'], unique=False - ) - - op.create_table( - 'run_node', - sa.Column('run_id', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False), - sa.Column('node', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False), - sa.Column('status', sqlmodel.sql.sqltypes.AutoString(length=16), nullable=False), - sa.Column('attempt', sa.Integer(), nullable=False), - sa.Column('started_at', sa.DateTime(timezone=True), nullable=True), - sa.Column('duration_ms', sa.Float(), nullable=False), - sa.Column('worker', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False), - sa.Column('error', sqlmodel.sql.sqltypes.AutoString(), nullable=False), - sa.Column('logs', sqlmodel.sql.sqltypes.AutoString(), nullable=False), - sa.Column( - 'cache_key', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False - ), - sa.PrimaryKeyConstraint('run_id', 'node'), - ) - op.create_index( - op.f('ix_run_node_cache_key'), 'run_node', ['cache_key'], unique=False - ) - - op.create_table( - 'run_metric', - sa.Column('run_id', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False), - sa.Column('name', sqlmodel.sql.sqltypes.AutoString(length=128), nullable=False), - sa.Column('step', sa.Integer(), nullable=False), - sa.Column('node', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False), - sa.Column('ts', sa.Float(), nullable=False), - sa.Column('value', sa.Float(), nullable=False), - sa.PrimaryKeyConstraint('run_id', 'name', 'step'), - ) - - op.create_table( - 'run_artifact', - sa.Column('run_id', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False), - sa.Column('name', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False), - sa.Column('node', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False), - sa.Column('digest', sqlmodel.sql.sqltypes.AutoString(length=71), nullable=False), - sa.Column('size', sa.BigInteger(), nullable=False), - sa.Column( - 'media_type', sqlmodel.sql.sqltypes.AutoString(length=128), nullable=False - ), - sa.Column('created_at', sa.DateTime(timezone=True), nullable=False), - sa.PrimaryKeyConstraint('run_id', 'name'), - ) - op.create_index( - op.f('ix_run_artifact_digest'), 'run_artifact', ['digest'], unique=False - ) - - -def downgrade(): - op.drop_index(op.f('ix_run_artifact_digest'), table_name='run_artifact') - op.drop_table('run_artifact') - op.drop_table('run_metric') - op.drop_index(op.f('ix_run_node_cache_key'), table_name='run_node') - op.drop_table('run_node') - op.drop_index(op.f('ix_run_params_digest'), table_name='run') - op.drop_index(op.f('ix_run_created_at'), table_name='run') - op.drop_index(op.f('ix_run_group_id'), table_name='run') - op.drop_index(op.f('ix_run_status'), table_name='run') - op.drop_index(op.f('ix_run_flow'), table_name='run') - op.drop_table('run') diff --git a/backend/fluksio/alembic/versions/b7c41d2f8a30_drop_item_table.py b/backend/fluksio/alembic/versions/b7c41d2f8a30_drop_item_table.py deleted file mode 100644 index 372aeed..0000000 --- a/backend/fluksio/alembic/versions/b7c41d2f8a30_drop_item_table.py +++ /dev/null @@ -1,40 +0,0 @@ -"""Drop the item table - -The template's example model. Flows are persisted as files in their own git -repository, so nothing in the app uses it any more. - -Revision ID: b7c41d2f8a30 -Revises: fe56fa70289e -Create Date: 2026-08-15 - -""" - -import sqlalchemy as sa -import sqlmodel.sql.sqltypes -from alembic import op - -revision = "b7c41d2f8a30" -down_revision = "fe56fa70289e" -branch_labels = None -depends_on = None - - -def upgrade(): - op.drop_table("item") - - -def downgrade(): - op.create_table( - "item", - sa.Column("id", sa.Uuid(), nullable=False), - sa.Column( - "title", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False - ), - sa.Column( - "description", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=True - ), - sa.Column("created_at", sa.DateTime(timezone=True), nullable=True), - sa.Column("owner_id", sa.Uuid(), nullable=False), - sa.ForeignKeyConstraint(["owner_id"], ["user.id"], ondelete="CASCADE"), - sa.PrimaryKeyConstraint("id"), - ) diff --git a/backend/fluksio/alembic/versions/c4e8b2170f93_add_portal_sub_to_user.py b/backend/fluksio/alembic/versions/c4e8b2170f93_add_portal_sub_to_user.py deleted file mode 100644 index e618bbd..0000000 --- a/backend/fluksio/alembic/versions/c4e8b2170f93_add_portal_sub_to_user.py +++ /dev/null @@ -1,35 +0,0 @@ -"""Map local accounts to the portal identities they stand for - -A portal session used to act as whoever enrolled this installation. It now -names a person, and this column is where that name meets a local account: set -for the enrolling superuser at enrolment, and for every remote user a superuser -admits afterwards. Unique, because two local accounts standing for the same -portal identity would make the lookup a coin toss. - -Revision ID: c4e8b2170f93 -Revises: a3f1c07b52d9 -Create Date: 2026-08-21 10:42:07.512004 - -""" -import sqlalchemy as sa -import sqlmodel.sql.sqltypes -from alembic import op - -# revision identifiers, used by Alembic. -revision = 'c4e8b2170f93' -down_revision = 'a3f1c07b52d9' -branch_labels = None -depends_on = None - - -def upgrade(): - op.add_column( - 'user', - sa.Column('portal_sub', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=True), - ) - op.create_index(op.f('ix_user_portal_sub'), 'user', ['portal_sub'], unique=True) - - -def downgrade(): - op.drop_index(op.f('ix_user_portal_sub'), table_name='user') - op.drop_column('user', 'portal_sub') diff --git a/backend/fluksio/alembic/versions/d98dd8ec85a3_edit_replace_id_integers_in_all_models_.py b/backend/fluksio/alembic/versions/d98dd8ec85a3_edit_replace_id_integers_in_all_models_.py deleted file mode 100755 index 37af1fa..0000000 --- a/backend/fluksio/alembic/versions/d98dd8ec85a3_edit_replace_id_integers_in_all_models_.py +++ /dev/null @@ -1,90 +0,0 @@ -"""Edit replace id integers in all models to use UUID instead - -Revision ID: d98dd8ec85a3 -Revises: 9c0a54914c78 -Create Date: 2024-07-19 04:08:04.000976 - -""" -from alembic import op -import sqlalchemy as sa -import sqlmodel.sql.sqltypes -from sqlalchemy.dialects import postgresql - - -# revision identifiers, used by Alembic. -revision = 'd98dd8ec85a3' -down_revision = '9c0a54914c78' -branch_labels = None -depends_on = None - - -def upgrade(): - # Ensure uuid-ossp extension is available - op.execute('CREATE EXTENSION IF NOT EXISTS "uuid-ossp"') - - # Create a new UUID column with a default UUID value - op.add_column('user', sa.Column('new_id', postgresql.UUID(as_uuid=True), default=sa.text('uuid_generate_v4()'))) - op.add_column('item', sa.Column('new_id', postgresql.UUID(as_uuid=True), default=sa.text('uuid_generate_v4()'))) - op.add_column('item', sa.Column('new_owner_id', postgresql.UUID(as_uuid=True), nullable=True)) - - # Populate the new columns with UUIDs - op.execute('UPDATE "user" SET new_id = uuid_generate_v4()') - op.execute('UPDATE item SET new_id = uuid_generate_v4()') - op.execute('UPDATE item SET new_owner_id = (SELECT new_id FROM "user" WHERE "user".id = item.owner_id)') - - # Set the new_id as not nullable - op.alter_column('user', 'new_id', nullable=False) - op.alter_column('item', 'new_id', nullable=False) - - # Drop old columns and rename new columns - op.drop_constraint('item_owner_id_fkey', 'item', type_='foreignkey') - op.drop_column('item', 'owner_id') - op.alter_column('item', 'new_owner_id', new_column_name='owner_id') - - op.drop_column('user', 'id') - op.alter_column('user', 'new_id', new_column_name='id') - - op.drop_column('item', 'id') - op.alter_column('item', 'new_id', new_column_name='id') - - # Create primary key constraint - op.create_primary_key('user_pkey', 'user', ['id']) - op.create_primary_key('item_pkey', 'item', ['id']) - - # Recreate foreign key constraint - op.create_foreign_key('item_owner_id_fkey', 'item', 'user', ['owner_id'], ['id']) - -def downgrade(): - # Reverse the upgrade process - op.add_column('user', sa.Column('old_id', sa.Integer, autoincrement=True)) - op.add_column('item', sa.Column('old_id', sa.Integer, autoincrement=True)) - op.add_column('item', sa.Column('old_owner_id', sa.Integer, nullable=True)) - - # Populate the old columns with default values - # Generate sequences for the integer IDs if not exist - op.execute('CREATE SEQUENCE IF NOT EXISTS user_id_seq AS INTEGER OWNED BY "user".old_id') - op.execute('CREATE SEQUENCE IF NOT EXISTS item_id_seq AS INTEGER OWNED BY item.old_id') - - op.execute('SELECT setval(\'user_id_seq\', COALESCE((SELECT MAX(old_id) + 1 FROM "user"), 1), false)') - op.execute('SELECT setval(\'item_id_seq\', COALESCE((SELECT MAX(old_id) + 1 FROM item), 1), false)') - - op.execute('UPDATE "user" SET old_id = nextval(\'user_id_seq\')') - op.execute('UPDATE item SET old_id = nextval(\'item_id_seq\'), old_owner_id = (SELECT old_id FROM "user" WHERE "user".id = item.owner_id)') - - # Drop new columns and rename old columns back - op.drop_constraint('item_owner_id_fkey', 'item', type_='foreignkey') - op.drop_column('item', 'owner_id') - op.alter_column('item', 'old_owner_id', new_column_name='owner_id') - - op.drop_column('user', 'id') - op.alter_column('user', 'old_id', new_column_name='id') - - op.drop_column('item', 'id') - op.alter_column('item', 'old_id', new_column_name='id') - - # Create primary key constraint - op.create_primary_key('user_pkey', 'user', ['id']) - op.create_primary_key('item_pkey', 'item', ['id']) - - # Recreate foreign key constraint - op.create_foreign_key('item_owner_id_fkey', 'item', 'user', ['owner_id'], ['id']) diff --git a/backend/fluksio/alembic/versions/e2412789c190_initialize_models.py b/backend/fluksio/alembic/versions/e2412789c190_initialize_models.py deleted file mode 100644 index 7529ea9..0000000 --- a/backend/fluksio/alembic/versions/e2412789c190_initialize_models.py +++ /dev/null @@ -1,54 +0,0 @@ -"""Initialize models - -Revision ID: e2412789c190 -Revises: -Create Date: 2023-11-24 22:55:43.195942 - -""" -import sqlalchemy as sa -import sqlmodel.sql.sqltypes -from alembic import op - -# revision identifiers, used by Alembic. -revision = "e2412789c190" -down_revision = None -branch_labels = None -depends_on = None - - -def upgrade(): - # ### commands auto generated by Alembic - please adjust! ### - op.create_table( - "user", - sa.Column("email", sqlmodel.sql.sqltypes.AutoString(), nullable=False), - sa.Column("is_active", sa.Boolean(), nullable=False), - sa.Column("is_superuser", sa.Boolean(), nullable=False), - sa.Column("full_name", sqlmodel.sql.sqltypes.AutoString(), nullable=True), - sa.Column("id", sa.Integer(), nullable=False), - sa.Column( - "hashed_password", sqlmodel.sql.sqltypes.AutoString(), nullable=False - ), - sa.PrimaryKeyConstraint("id"), - ) - op.create_index(op.f("ix_user_email"), "user", ["email"], unique=True) - op.create_table( - "item", - sa.Column("description", sqlmodel.sql.sqltypes.AutoString(), nullable=True), - sa.Column("id", sa.Integer(), nullable=False), - sa.Column("title", sqlmodel.sql.sqltypes.AutoString(), nullable=False), - sa.Column("owner_id", sa.Integer(), nullable=False), - sa.ForeignKeyConstraint( - ["owner_id"], - ["user.id"], - ), - sa.PrimaryKeyConstraint("id"), - ) - # ### end Alembic commands ### - - -def downgrade(): - # ### commands auto generated by Alembic - please adjust! ### - op.drop_table("item") - op.drop_index(op.f("ix_user_email"), table_name="user") - op.drop_table("user") - # ### end Alembic commands ### diff --git a/backend/fluksio/alembic/versions/ee1b4b4426a3_baseline_schema.py b/backend/fluksio/alembic/versions/ee1b4b4426a3_baseline_schema.py new file mode 100644 index 0000000..e583d87 --- /dev/null +++ b/backend/fluksio/alembic/versions/ee1b4b4426a3_baseline_schema.py @@ -0,0 +1,337 @@ +"""baseline schema + +The schema as it stands, in one portable revision. The ten that came before +built it up on Postgres — with `uuid-ossp`, sequences and `setval` in them — +and none of them could run on SQLite, which is now the default. The history +they recorded is in git; what a database needs is the shape. + +Revision ID: ee1b4b4426a3 +Revises: +Create Date: 2026-08-21 + +""" + +import sqlalchemy as sa +import sqlmodel.sql.sqltypes +from alembic import op + +# revision identifiers, used by Alembic. +revision = "ee1b4b4426a3" +down_revision = None +branch_labels = None +depends_on = None + + +def upgrade(): + op.create_table( + "engine_event", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("ts", sa.DateTime(timezone=True), nullable=False), + sa.Column("type", sqlmodel.sql.sqltypes.AutoString(length=32), nullable=False), + sa.Column("flow", sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column("node", sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column("detail", sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column("actor", sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.PrimaryKeyConstraint("id"), + ) + with op.batch_alter_table("engine_event", schema=None) as batch_op: + batch_op.create_index(batch_op.f("ix_engine_event_ts"), ["ts"], unique=False) + batch_op.create_index( + batch_op.f("ix_engine_event_type"), ["type"], unique=False + ) + + op.create_table( + "flow_run", + sa.Column("id", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False), + sa.Column("flow", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False), + sa.Column("source", sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("status", sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column("nodes", sa.Integer(), nullable=False), + sa.Column("errors", sa.Integer(), nullable=False), + sa.Column("duration_ms", sa.Float(), nullable=False), + sa.Column("deliveries", sa.Integer(), nullable=False), + sa.PrimaryKeyConstraint("id"), + ) + with op.batch_alter_table("flow_run", schema=None) as batch_op: + batch_op.create_index(batch_op.f("ix_flow_run_flow"), ["flow"], unique=False) + batch_op.create_index( + batch_op.f("ix_flow_run_started_at"), ["started_at"], unique=False + ) + + op.create_table( + "metric_minute", + sa.Column("flow", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False), + sa.Column("node", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False), + sa.Column("bucket", sa.DateTime(timezone=True), nullable=False), + sa.Column("executions", sa.Integer(), nullable=False), + sa.Column("errors", sa.Integer(), nullable=False), + sa.Column("messages", sa.Integer(), nullable=False), + sa.Column("duration_sum_ms", sa.Float(), nullable=False), + sa.Column("duration_max_ms", sa.Float(), nullable=False), + sa.Column("lag_sum_ms", sa.Float(), nullable=False), + sa.Column("lag_max_ms", sa.Float(), nullable=False), + sa.Column("items", sa.Integer(), nullable=False), + sa.PrimaryKeyConstraint("flow", "node", "bucket"), + ) + with op.batch_alter_table("metric_minute", schema=None) as batch_op: + batch_op.create_index( + batch_op.f("ix_metric_minute_bucket"), ["bucket"], unique=False + ) + + op.create_table( + "oauth_client", + sa.Column("id", sa.Uuid(), nullable=False), + sa.Column( + "client_name", sqlmodel.sql.sqltypes.AutoString(length=128), nullable=False + ), + sa.Column("redirect_uris", sa.JSON(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("id"), + ) + op.create_table( + "run", + sa.Column("id", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False), + sa.Column("flow", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False), + sa.Column("flow_version", sa.Integer(), nullable=False), + sa.Column( + "commit", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False + ), + sa.Column("params", sa.JSON(), nullable=True), + sa.Column( + "params_digest", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False + ), + sa.Column("seed", sa.Integer(), nullable=True), + sa.Column( + "group_id", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=True + ), + sa.Column( + "parent_id", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=True + ), + sa.Column("cause", sqlmodel.sql.sqltypes.AutoString(length=32), nullable=False), + sa.Column( + "status", sqlmodel.sql.sqltypes.AutoString(length=16), nullable=False + ), + sa.Column( + "status_reason", + sqlmodel.sql.sqltypes.AutoString(length=1024), + nullable=False, + ), + sa.Column("labels", sa.JSON(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("duration_ms", sa.Float(), nullable=False), + sa.Column("result", sa.JSON(), nullable=True), + sa.Column( + "engine", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False + ), + sa.Column("lease_at", sa.DateTime(timezone=True), nullable=True), + sa.Column( + "actor", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False + ), + sa.PrimaryKeyConstraint("id"), + ) + with op.batch_alter_table("run", schema=None) as batch_op: + batch_op.create_index( + batch_op.f("ix_run_created_at"), ["created_at"], unique=False + ) + batch_op.create_index(batch_op.f("ix_run_flow"), ["flow"], unique=False) + batch_op.create_index(batch_op.f("ix_run_group_id"), ["group_id"], unique=False) + batch_op.create_index( + batch_op.f("ix_run_params_digest"), ["params_digest"], unique=False + ) + batch_op.create_index(batch_op.f("ix_run_status"), ["status"], unique=False) + + op.create_table( + "run_artifact", + sa.Column( + "run_id", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False + ), + sa.Column("name", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False), + sa.Column("node", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False), + sa.Column( + "digest", sqlmodel.sql.sqltypes.AutoString(length=71), nullable=False + ), + sa.Column("size", sa.Integer(), nullable=False), + sa.Column( + "media_type", sqlmodel.sql.sqltypes.AutoString(length=128), nullable=False + ), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("run_id", "name"), + ) + with op.batch_alter_table("run_artifact", schema=None) as batch_op: + batch_op.create_index( + batch_op.f("ix_run_artifact_digest"), ["digest"], unique=False + ) + + op.create_table( + "run_metric", + sa.Column( + "run_id", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False + ), + sa.Column("name", sqlmodel.sql.sqltypes.AutoString(length=128), nullable=False), + sa.Column("step", sa.Integer(), nullable=False), + sa.Column("node", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False), + sa.Column("ts", sa.Float(), nullable=False), + sa.Column("value", sa.Float(), nullable=False), + sa.PrimaryKeyConstraint("run_id", "name", "step"), + ) + op.create_table( + "run_node", + sa.Column( + "run_id", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False + ), + sa.Column("node", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False), + sa.Column( + "status", sqlmodel.sql.sqltypes.AutoString(length=16), nullable=False + ), + sa.Column("attempt", sa.Integer(), nullable=False), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("duration_ms", sa.Float(), nullable=False), + sa.Column( + "worker", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False + ), + sa.Column("error", sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column("logs", sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column( + "cache_key", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False + ), + sa.PrimaryKeyConstraint("run_id", "node"), + ) + with op.batch_alter_table("run_node", schema=None) as batch_op: + batch_op.create_index( + batch_op.f("ix_run_node_cache_key"), ["cache_key"], unique=False + ) + + op.create_table( + "user", + sa.Column( + "email", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False + ), + sa.Column("is_active", sa.Boolean(), nullable=False), + sa.Column("is_superuser", sa.Boolean(), nullable=False), + sa.Column( + "full_name", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=True + ), + sa.Column("id", sa.Uuid(), nullable=False), + sa.Column( + "hashed_password", sqlmodel.sql.sqltypes.AutoString(), nullable=False + ), + sa.Column( + "portal_sub", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=True + ), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=True), + sa.PrimaryKeyConstraint("id"), + ) + with op.batch_alter_table("user", schema=None) as batch_op: + batch_op.create_index(batch_op.f("ix_user_email"), ["email"], unique=True) + batch_op.create_index( + batch_op.f("ix_user_portal_sub"), ["portal_sub"], unique=True + ) + + op.create_table( + "oauth_authorization_code", + sa.Column("id", sa.Uuid(), nullable=False), + sa.Column( + "code_hash", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False + ), + sa.Column("client_id", sa.Uuid(), nullable=False), + sa.Column("user_id", sa.Uuid(), nullable=False), + sa.Column( + "redirect_uri", + sqlmodel.sql.sqltypes.AutoString(length=2048), + nullable=False, + ), + sa.Column( + "code_challenge", + sqlmodel.sql.sqltypes.AutoString(length=128), + nullable=False, + ), + sa.Column( + "resource", sqlmodel.sql.sqltypes.AutoString(length=2048), nullable=True + ), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("used_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("refresh_token_id", sa.Uuid(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint(["client_id"], ["oauth_client.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["user_id"], ["user.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + ) + with op.batch_alter_table("oauth_authorization_code", schema=None) as batch_op: + batch_op.create_index( + batch_op.f("ix_oauth_authorization_code_code_hash"), + ["code_hash"], + unique=True, + ) + + op.create_table( + "oauth_refresh_token", + sa.Column("id", sa.Uuid(), nullable=False), + sa.Column( + "token_hash", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False + ), + sa.Column("client_id", sa.Uuid(), nullable=False), + sa.Column("user_id", sa.Uuid(), nullable=False), + sa.Column("family_id", sa.Uuid(), nullable=False), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("revoked", sa.Boolean(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint(["client_id"], ["oauth_client.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["user_id"], ["user.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + ) + with op.batch_alter_table("oauth_refresh_token", schema=None) as batch_op: + batch_op.create_index( + batch_op.f("ix_oauth_refresh_token_token_hash"), ["token_hash"], unique=True + ) + + +def downgrade(): + with op.batch_alter_table("oauth_refresh_token", schema=None) as batch_op: + batch_op.drop_index(batch_op.f("ix_oauth_refresh_token_token_hash")) + + op.drop_table("oauth_refresh_token") + with op.batch_alter_table("oauth_authorization_code", schema=None) as batch_op: + batch_op.drop_index(batch_op.f("ix_oauth_authorization_code_code_hash")) + + op.drop_table("oauth_authorization_code") + with op.batch_alter_table("user", schema=None) as batch_op: + batch_op.drop_index(batch_op.f("ix_user_portal_sub")) + batch_op.drop_index(batch_op.f("ix_user_email")) + + op.drop_table("user") + with op.batch_alter_table("run_node", schema=None) as batch_op: + batch_op.drop_index(batch_op.f("ix_run_node_cache_key")) + + op.drop_table("run_node") + op.drop_table("run_metric") + with op.batch_alter_table("run_artifact", schema=None) as batch_op: + batch_op.drop_index(batch_op.f("ix_run_artifact_digest")) + + op.drop_table("run_artifact") + with op.batch_alter_table("run", schema=None) as batch_op: + batch_op.drop_index(batch_op.f("ix_run_status")) + batch_op.drop_index(batch_op.f("ix_run_params_digest")) + batch_op.drop_index(batch_op.f("ix_run_group_id")) + batch_op.drop_index(batch_op.f("ix_run_flow")) + batch_op.drop_index(batch_op.f("ix_run_created_at")) + + op.drop_table("run") + op.drop_table("oauth_client") + with op.batch_alter_table("metric_minute", schema=None) as batch_op: + batch_op.drop_index(batch_op.f("ix_metric_minute_bucket")) + + op.drop_table("metric_minute") + with op.batch_alter_table("flow_run", schema=None) as batch_op: + batch_op.drop_index(batch_op.f("ix_flow_run_started_at")) + batch_op.drop_index(batch_op.f("ix_flow_run_flow")) + + op.drop_table("flow_run") + with op.batch_alter_table("engine_event", schema=None) as batch_op: + batch_op.drop_index(batch_op.f("ix_engine_event_type")) + batch_op.drop_index(batch_op.f("ix_engine_event_ts")) + + op.drop_table("engine_event") diff --git a/backend/fluksio/alembic/versions/fe56fa70289e_add_created_at_to_user_and_item.py b/backend/fluksio/alembic/versions/fe56fa70289e_add_created_at_to_user_and_item.py deleted file mode 100644 index 3e15754..0000000 --- a/backend/fluksio/alembic/versions/fe56fa70289e_add_created_at_to_user_and_item.py +++ /dev/null @@ -1,31 +0,0 @@ -"""Add created_at to User and Item - -Revision ID: fe56fa70289e -Revises: 1a31ce608336 -Create Date: 2026-01-23 15:50:37.171462 - -""" -from alembic import op -import sqlalchemy as sa -import sqlmodel.sql.sqltypes - - -# revision identifiers, used by Alembic. -revision = 'fe56fa70289e' -down_revision = '1a31ce608336' -branch_labels = None -depends_on = None - - -def upgrade(): - # ### commands auto generated by Alembic - please adjust! ### - op.add_column('item', sa.Column('created_at', sa.DateTime(timezone=True), nullable=True)) - op.add_column('user', sa.Column('created_at', sa.DateTime(timezone=True), nullable=True)) - # ### end Alembic commands ### - - -def downgrade(): - # ### commands auto generated by Alembic - please adjust! ### - op.drop_column('user', 'created_at') - op.drop_column('item', 'created_at') - # ### end Alembic commands ### diff --git a/backend/fluksio/api/deps.py b/backend/fluksio/api/deps.py index 96af00e..c433c29 100644 --- a/backend/fluksio/api/deps.py +++ b/backend/fluksio/api/deps.py @@ -1,3 +1,4 @@ +import uuid from collections.abc import Generator from typing import Annotated, Any @@ -157,7 +158,16 @@ def _user_for(session: Session, token_data: TokenPayload) -> User | None: return session.exec( select(User).where(User.portal_sub == token_data.portal_sub) ).first() - return session.get(User, token_data.sub) if token_data.sub else None + if not token_data.sub: + return None + try: + # The subject is a string on the wire. Postgres cast it on the way in; + # nothing else does, so parse it here rather than hand a driver a + # string where it wants a UUID. + user_id = uuid.UUID(token_data.sub) + except ValueError: + return None + return session.get(User, user_id) def user_from_token( diff --git a/backend/fluksio/api/routes/observability.py b/backend/fluksio/api/routes/observability.py index feb26e1..59558e4 100644 --- a/backend/fluksio/api/routes/observability.py +++ b/backend/fluksio/api/routes/observability.py @@ -13,7 +13,7 @@ from typing import Any, Literal from fastapi import APIRouter, Depends, Request from fastapi.concurrency import run_in_threadpool from pydantic import BaseModel -from sqlalchemy import ColumnElement, DateTime, Interval, cast, func, literal +from sqlalchemy import ColumnElement, Integer, cast, func from sqlalchemy import select as sa_select from sqlmodel import col, select @@ -31,9 +31,15 @@ router = APIRouter( #: How many slices a per-flow sparkline is folded into. SPARK_SLICES = 60 -#: The origin fixed-stride slots are aligned to, which is the alignment the -#: fold used to get from ``stamp - stamp % bucket_s``. -EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc) + +def _epoch(column: Any) -> ColumnElement[int]: + """Seconds since 1970, as an integer the database can bin on. + + Binning is integer arithmetic on this rather than a dialect's own date + function: buckets are whole minutes, so the cast is exact, and the + expression is the same everywhere. + """ + return cast(func.extract("epoch", column), Integer) class HealthSummary(BaseModel): @@ -203,16 +209,10 @@ def read_timeseries( ) -> Any: """Executions, errors and timings over time, summed across nodes.""" hours = _window_hours(hours) - # Postgres does the fold: a week of minute rows per node used to cross the - # wire on every poll, and only the slices need to. The casts are load - # bearing — date_bin() is overloaded on timestamp and timestamptz, and an - # untyped bind parameter leaves the call ambiguous. - stride = timedelta(seconds=max(60, bucket_s)) - slot = func.date_bin( - cast(literal(stride), Interval), - col(MetricBucket.bucket), - cast(literal(EPOCH), DateTime(timezone=True)), - ).label("slot") + # The database does the fold: a week of minute rows per node used to cross + # the wire on every poll, and only the slices need to. + stride = max(60, bucket_s) + slot = (_epoch(col(MetricBucket.bucket)) // stride * stride).label("slot") statement = sa_select( slot, func.sum(col(MetricBucket.executions)).label("executions"), @@ -230,7 +230,7 @@ def read_timeseries( return [ SeriesPoint( - ts=row.slot.timestamp(), + ts=float(row.slot), executions=int(row.executions), errors=int(row.errors), messages=int(row.messages), @@ -253,10 +253,10 @@ def read_flow_rollups(session: SessionDep, hours: int = 24) -> Any: # Binned to the sparkline slice rather than the minute, so a flow costs at # most SPARK_SLICES rows however long the window is. The slice is the # window over SPARK_SLICES, which for whole hours is whole minutes. - slot = func.date_bin( - cast(literal(timedelta(minutes=hours)), Interval), - col(MetricBucket.bucket), - cast(literal(since), DateTime(timezone=True)), + stride = hours * 60 + origin = int(start) + slot = ( + (_epoch(col(MetricBucket.bucket)) - origin) // stride * stride + origin ).label("slot") statement = ( sa_select( @@ -298,7 +298,7 @@ def read_flow_rollups(session: SessionDep, hours: int = 24) -> Any: # clamp holds the bucket landing exactly on the far edge in range. index = min( SPARK_SLICES - 1, - max(0, round((row.slot.timestamp() - start) / window * SPARK_SLICES)), + max(0, round((float(row.slot) - start) / window * SPARK_SLICES)), ) entry["spark"][index] += row.executions diff --git a/backend/fluksio/backend_pre_start.py b/backend/fluksio/backend_pre_start.py deleted file mode 100644 index 8f68fd1..0000000 --- a/backend/fluksio/backend_pre_start.py +++ /dev/null @@ -1,39 +0,0 @@ -import logging - -from sqlalchemy import Engine -from sqlmodel import Session, select -from tenacity import after_log, before_log, retry, stop_after_attempt, wait_fixed - -from fluksio.core.db import engine - -logging.basicConfig(level=logging.INFO) -logger = logging.getLogger(__name__) - -max_tries = 60 * 5 # 5 minutes -wait_seconds = 1 - - -@retry( - stop=stop_after_attempt(max_tries), - wait=wait_fixed(wait_seconds), - before=before_log(logger, logging.INFO), - after=after_log(logger, logging.WARN), -) -def init(db_engine: Engine) -> None: - try: - with Session(db_engine) as session: - # Try to create session to check if DB is awake - session.exec(select(1)) - except Exception as e: - logger.error(e) - raise e - - -def main() -> None: - logger.info("Initializing service") - init(engine) - logger.info("Service finished initializing") - - -if __name__ == "__main__": - main() diff --git a/backend/fluksio/cloud/connector.py b/backend/fluksio/cloud/connector.py index e98a7ba..48b33b4 100644 --- a/backend/fluksio/cloud/connector.py +++ b/backend/fluksio/cloud/connector.py @@ -20,6 +20,7 @@ import base64 import contextlib import logging import time +import uuid from datetime import timedelta from typing import Any @@ -170,7 +171,7 @@ class CloudConnector: select(User).where(User.portal_sub == owner_id) ).first(): return - user = session.get(User, config.local_user_id) + user = session.get(User, uuid.UUID(config.local_user_id)) if user is None or user.portal_sub: return user.portal_sub = owner_id diff --git a/backend/fluksio/core/config.py b/backend/fluksio/core/config.py index 1c8a314..35723db 100644 --- a/backend/fluksio/core/config.py +++ b/backend/fluksio/core/config.py @@ -1,3 +1,4 @@ +import os import secrets import warnings from pathlib import Path @@ -8,13 +9,24 @@ from pydantic import ( BeforeValidator, EmailStr, HttpUrl, - PostgresDsn, computed_field, model_validator, ) from pydantic_settings import BaseSettings, SettingsConfigDict from typing_extensions import Self +#: Everything the engine keeps on disk, relative to :attr:`Settings.DATA_DIR`. +#: One setting to move the lot; each still overridable on its own, which is +#: what the container images do. +DERIVED_PATHS = { + "FLOWS_DIR": "flows", + "SECRETS_FILE": "secrets.enc", + "ALERTS_FILE": "alerts.json", + "PANELS_FILE": "panels.json", + "OAUTH_PRIVATE_KEY_FILE": "oauth-key.pem", + "CLOUD_CONFIG_FILE": "cloud.json", +} + def parse_cors(v: Any) -> list[str] | str: if isinstance(v, str) and not v.startswith("["): @@ -26,8 +38,10 @@ def parse_cors(v: Any) -> list[str] | str: class Settings(BaseSettings): model_config = SettingsConfigDict( - # Use top level .env file (one level above ./backend/) - env_file="../.env", + # The stack's own file, one level above ./backend/. An installed + # `fluksio` has no such tree, so its CLI points this at the data + # directory instead — and at nothing it might find in the cwd. + env_file=os.environ.get("FLUKSIO_ENV_FILE", "../.env"), env_ignore_empty=True, extra="ignore", ) @@ -38,6 +52,14 @@ class Settings(BaseSettings): FRONTEND_HOST: str = "http://localhost:5173" ENVIRONMENT: Literal["local", "staging", "production"] = "local" + #: Everything this installation keeps: the database, the flow repository, + #: secrets, artifacts and the user venv. The paths below derive from it + #: unless they are set explicitly. + DATA_DIR: Path = Path("flow-data") + #: Any SQLAlchemy URL. The default puts SQLite in the data directory, which + #: is what makes `fluksio serve` need no infrastructure at all. + DATABASE_URL: str | None = None + # Flows live on disk as a git repository; secrets stay outside it. FLOWS_DIR: Path = Path("flow-data/flows") SECRETS_FILE: Path = Path("flow-data/secrets.enc") @@ -80,6 +102,22 @@ class Settings(BaseSettings): list[AnyUrl] | str, BeforeValidator(parse_cors) ] = [] + @model_validator(mode="before") + @classmethod + def _derive_data_paths(cls, data: Any) -> Any: + """Put every stored thing under ``DATA_DIR`` unless it was named. + + ``setdefault``, so the container images keep their explicit ``/data`` + paths and a checkout keeps ``flow-data/``. + """ + if not isinstance(data, dict): + return data + base = Path(str(data.get("DATA_DIR", "flow-data"))).expanduser() + data["DATA_DIR"] = base + for key, name in DERIVED_PATHS.items(): + data.setdefault(key, base / name) + return data + @computed_field # type: ignore[prop-decorator] @property def oauth_issuer(self) -> str: @@ -105,25 +143,21 @@ class Settings(BaseSettings): self.FRONTEND_HOST ] - PROJECT_NAME: str + PROJECT_NAME: str = "Fluksio" SENTRY_DSN: HttpUrl | None = None - POSTGRES_SERVER: str - POSTGRES_PORT: int = 5432 - POSTGRES_USER: str - POSTGRES_PASSWORD: str = "" - POSTGRES_DB: str = "" @computed_field # type: ignore[prop-decorator] @property - def SQLALCHEMY_DATABASE_URI(self) -> PostgresDsn: - return PostgresDsn.build( - scheme="postgresql+psycopg", - username=self.POSTGRES_USER, - password=self.POSTGRES_PASSWORD, - host=self.POSTGRES_SERVER, - port=self.POSTGRES_PORT, - path=self.POSTGRES_DB, - ) + def SQLALCHEMY_DATABASE_URI(self) -> str: + """SQLite in the data directory, unless a URL says otherwise. + + One engine process owns this database — the same reason the image runs + a single uvicorn worker — so a file beside the flows is the honest + shape for it, and needs nothing running to be one. + """ + if self.DATABASE_URL: + return self.DATABASE_URL + return f"sqlite:///{(self.DATA_DIR / 'fluksio.db').expanduser().resolve()}" SMTP_TLS: bool = True SMTP_SSL: bool = False @@ -148,8 +182,10 @@ class Settings(BaseSettings): return bool(self.SMTP_HOST and self.EMAILS_FROM_EMAIL) EMAIL_TEST_USER: EmailStr = "test@example.com" - FIRST_SUPERUSER: EmailStr - FIRST_SUPERUSER_PASSWORD: str + # Absent means "the CLI will make one on first run" — a pip install is not + # asked for two environment variables before it can start. + FIRST_SUPERUSER: EmailStr | None = None + FIRST_SUPERUSER_PASSWORD: str | None = None def _check_default_secret(self, var_name: str, value: str | None) -> None: if value == "changethis": @@ -165,7 +201,6 @@ class Settings(BaseSettings): @model_validator(mode="after") def _enforce_non_default_secrets(self) -> Self: self._check_default_secret("SECRET_KEY", self.SECRET_KEY) - self._check_default_secret("POSTGRES_PASSWORD", self.POSTGRES_PASSWORD) self._check_default_secret( "FIRST_SUPERUSER_PASSWORD", self.FIRST_SUPERUSER_PASSWORD ) @@ -173,4 +208,6 @@ class Settings(BaseSettings): return self -settings = Settings() # type: ignore +# No arguments and no required environment: a fresh install boots on the +# defaults above, into a data directory of its own. +settings = Settings() diff --git a/backend/fluksio/core/db.py b/backend/fluksio/core/db.py index b06deb3..9a0769c 100644 --- a/backend/fluksio/core/db.py +++ b/backend/fluksio/core/db.py @@ -1,28 +1,103 @@ -from sqlmodel import Session, create_engine, select +"""The database: one file beside the flows, or whatever ``DATABASE_URL`` says. + +SQLite is the default because one engine process owns this database. Node +execution, live values and the work queue never come here — they are the state +backend's and the bus's — so what lands in a transaction is the rollups a +minute at a time, a row per cascade, and the run history. WAL lets the readers +carry on while that single writer works. +""" + +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Any + +from alembic import command +from alembic.config import Config +from sqlalchemy import Engine, event, inspect, make_url +from sqlmodel import Session, SQLModel, create_engine, select from fluksio import crud from fluksio.core.config import settings from fluksio.models import User, UserCreate -# A connection idle across a Postgres restart is dead but still pooled; the -# pre-ping spends a round trip to find out instead of failing the request. -engine = create_engine(str(settings.SQLALCHEMY_DATABASE_URI), pool_pre_ping=True) +logger = logging.getLogger(__name__) -# make sure all SQLModel models are imported (app.models) before initializing DB -# otherwise, SQLModel might fail to initialize relationships properly -# for more details: https://github.com/fastapi/full-stack-fastapi-template/issues/28 +def make_engine(url: str) -> Engine: + """One engine for the process, configured for whichever dialect it names.""" + if make_url(url).get_backend_name() != "sqlite": + # A connection idle across a server restart is dead but still pooled; + # the pre-ping spends a round trip to find out instead of failing the + # request. + return create_engine(url, pool_pre_ping=True) + + database = make_url(url).database + if database and database != ":memory:": + Path(database).parent.mkdir(parents=True, exist_ok=True) + # The engine's threads — the run service, the metrics collector, the + # request pool — share this engine, so connections cross threads. Each one + # is still used by a single thread at a time; the pool sees to that. + return create_engine(url, connect_args={"check_same_thread": False}) + + +engine = make_engine(settings.SQLALCHEMY_DATABASE_URI) + + +@event.listens_for(engine, "connect") +def _sqlite_pragmas(dbapi_connection: Any, _record: Any) -> None: + """What makes concurrent readers, cascades and waiting-out a writer work.""" + if engine.dialect.name != "sqlite": + return + cursor = dbapi_connection.cursor() + # Readers do not block on the writer, which is the whole reason this is + # usable while a run is streaming metrics into it. + cursor.execute("PRAGMA journal_mode=WAL") + # Commit without waiting for the platter. A power cut can lose the last + # transactions; it cannot corrupt the file. + cursor.execute("PRAGMA synchronous=NORMAL") + # Off by default in SQLite, and `ondelete="CASCADE"` on the OAuth tables is + # the whole of revoking a client. + cursor.execute("PRAGMA foreign_keys=ON") + # A second writer waits rather than raising "database is locked". + cursor.execute("PRAGMA busy_timeout=30000") + cursor.close() + + +def _alembic_config(connection: Any) -> Config: + """Alembic driven from here, so an installed wheel needs no alembic.ini.""" + config = Config() + config.set_main_option( + "script_location", str(Path(__file__).parents[1] / "alembic") + ) + config.attributes["connection"] = connection + return config + + +def migrate(engine: Engine) -> None: + """Bring the schema up to date, creating it if there is nothing there.""" + with engine.begin() as connection: + config = _alembic_config(connection) + tables = inspect(connection).get_table_names() + if not tables: + # Nothing to upgrade from. Building the schema from the models is + # both faster and exactly what the revisions would have produced. + SQLModel.metadata.create_all(connection) + command.stamp(config, "head") + logger.info("created the database schema") + return + command.upgrade(config, "head") def init_db(session: Session) -> None: - # Tables should be created with Alembic migrations - # But if you don't want to use migrations, create - # the tables un-commenting the next lines - # from sqlmodel import SQLModel - - # This works because the models are already imported and registered from app.models - # SQLModel.metadata.create_all(engine) + """The first superuser, when the deployment names one. + A `fluksio serve` with no configuration makes its own and prints it once; + that path calls into `fluksio.core.bootstrap` instead of this. + """ + if not (settings.FIRST_SUPERUSER and settings.FIRST_SUPERUSER_PASSWORD): + return user = session.exec( select(User).where(User.email == settings.FIRST_SUPERUSER) ).first() @@ -33,3 +108,10 @@ def init_db(session: Session) -> None: is_superuser=True, ) user = crud.create_user(session=session, user_create=user_in) + + +def prepare(engine: Engine) -> None: + """Everything that has to be true before the app serves a request.""" + migrate(engine) + with Session(engine) as session: + init_db(session) diff --git a/backend/fluksio/core/types.py b/backend/fluksio/core/types.py new file mode 100644 index 0000000..8f30488 --- /dev/null +++ b/backend/fluksio/core/types.py @@ -0,0 +1,44 @@ +"""Column types the models share. + +Its own module because :mod:`fluksio.core.db` imports the models, so the +models cannot import from there. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any + +from sqlalchemy import DateTime, TypeDecorator + + +class UTCDateTime(TypeDecorator[datetime]): + """A timestamp that is still UTC-aware when it comes back. + + SQLite has no timezone-aware type: it stores what it is handed and returns + it naive, so a column read back would compare against ``now(timezone.utc)`` + with a TypeError, or — worse — serialise without an offset and be read as + local time by whoever gets the JSON. Every stored instant is converted to + UTC on the way in and labelled UTC on the way out, on every dialect. + """ + + impl = DateTime(timezone=True) + cache_ok = True + + def process_bind_param( + self, value: datetime | None, dialect: Any + ) -> datetime | None: + if value is None: + return None + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + def process_result_value( + self, value: datetime | None, dialect: Any + ) -> datetime | None: + if value is None: + return None + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) diff --git a/backend/fluksio/flow/metrics.py b/backend/fluksio/flow/metrics.py index 7319d27..0c63e3f 100644 --- a/backend/fluksio/flow/metrics.py +++ b/backend/fluksio/flow/metrics.py @@ -19,7 +19,7 @@ from datetime import datetime, timedelta, timezone from typing import Any from sqlalchemy import delete, func, update -from sqlalchemy.dialects.postgresql import insert +from sqlalchemy.dialects.sqlite import insert from sqlmodel import Session, col from fluksio.core.config import settings @@ -354,9 +354,9 @@ class MetricsCollector: for name in SUMMED } | { - name: func.greatest( - col(getattr(MetricBucket, name)), new[name] - ) + # `max` of two values, not the aggregate: SQLite's + # scalar form, which is what `greatest` is elsewhere. + name: func.max(col(getattr(MetricBucket, name)), new[name]) for name in MAXIMA }, ) diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index c6efa64..bca4411 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -39,7 +39,7 @@ from datetime import datetime, timedelta, timezone from typing import Any from sqlalchemy import update -from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.dialects.sqlite import insert as upsert from sqlmodel import Session, col, select from fluksio.core.db import engine as db_engine @@ -258,9 +258,7 @@ class MetricSink: def _write(self, rows: list[RunMetric]) -> None: try: with Session(db_engine) as session: - statement = pg_insert(RunMetric).values( - [row.model_dump() for row in rows] - ) + statement = upsert(RunMetric).values([row.model_dump() for row in rows]) session.exec( statement.on_conflict_do_update( index_elements=["run_id", "name", "step"], diff --git a/backend/fluksio/flow/workers.py b/backend/fluksio/flow/workers.py index 068e6ca..bf3b46b 100644 --- a/backend/fluksio/flow/workers.py +++ b/backend/fluksio/flow/workers.py @@ -42,7 +42,7 @@ COMPILE_TIMEOUT = 60.0 #: Environment the worker is not given. ``SECRET_KEY`` decrypts every stored #: secret, not only the ones bound to the node asking. -ENV_DENY_PREFIXES = ("POSTGRES_", "FIRST_SUPERUSER", "SENTRY_DSN") +ENV_DENY_PREFIXES = ("DATABASE_URL", "FIRST_SUPERUSER", "SENTRY_DSN") ENV_DENY_WORDS = ("PASSWORD", "SECRET", "TOKEN") diff --git a/backend/fluksio/initial_data.py b/backend/fluksio/initial_data.py deleted file mode 100644 index 0609d24..0000000 --- a/backend/fluksio/initial_data.py +++ /dev/null @@ -1,23 +0,0 @@ -import logging - -from sqlmodel import Session - -from fluksio.core.db import engine, init_db - -logging.basicConfig(level=logging.INFO) -logger = logging.getLogger(__name__) - - -def init() -> None: - with Session(engine) as session: - init_db(session) - - -def main() -> None: - logger.info("Creating initial data") - init() - logger.info("Initial data created") - - -if __name__ == "__main__": - main() diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index 174c58d..8cb0528 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -16,6 +16,8 @@ from fluksio.api.routes.alerts import read_config as read_alerts_config from fluksio.cloud import config as cloud_config from fluksio.core import security from fluksio.core.config import settings +from fluksio.core.db import engine as db_engine +from fluksio.core.db import prepare from fluksio.flow import logs, modules from fluksio.flow.alerts import AlertManager from fluksio.flow.artifacts import ArtifactStore @@ -88,6 +90,10 @@ def _mcp_sessions() -> AbstractAsyncContextManager[None]: @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[None]: """Start the flow engine alongside the API.""" + # The schema and the first superuser, before anything reads either. It is + # idempotent, so a deployment that ran this from its own prestart step + # pays a version check for it and nothing else. + await run_in_threadpool(prepare, db_engine) event_bus.bind(asyncio.get_running_loop()) # Node code is user code, and `print` is how it says things. logs.install() diff --git a/backend/fluksio/models.py b/backend/fluksio/models.py index 6bedd8a..7764a4c 100644 --- a/backend/fluksio/models.py +++ b/backend/fluksio/models.py @@ -3,9 +3,11 @@ from datetime import datetime, timezone from typing import Any from pydantic import EmailStr -from sqlalchemy import JSON, Column, DateTime +from sqlalchemy import JSON, Column from sqlmodel import Field, SQLModel +from fluksio.core.types import UTCDateTime + def get_datetime_utc() -> datetime: return datetime.now(timezone.utc) @@ -57,7 +59,7 @@ class User(UserBase, table=True): portal_sub: str | None = Field(default=None, max_length=64, unique=True, index=True) created_at: datetime | None = Field( default_factory=get_datetime_utc, - sa_type=DateTime(timezone=True), # type: ignore + sa_type=UTCDateTime, ) @@ -118,7 +120,9 @@ class OAuthClient(SQLModel, table=True): client_name: str = Field(max_length=128) redirect_uris: list[str] = Field(sa_column=Column(JSON), default_factory=list) created_at: datetime = Field( - default_factory=lambda: datetime.now(timezone.utc), nullable=False + default_factory=lambda: datetime.now(timezone.utc), + nullable=False, + sa_type=UTCDateTime, ) @@ -138,12 +142,14 @@ class OAuthAuthorizationCode(SQLModel, table=True): redirect_uri: str = Field(max_length=2048) code_challenge: str = Field(max_length=128) resource: str | None = Field(default=None, max_length=2048) - expires_at: datetime - used_at: datetime | None = None + expires_at: datetime = Field(sa_type=UTCDateTime) + used_at: datetime | None = Field(default=None, sa_type=UTCDateTime) #: The refresh token this code produced, so replaying the code can revoke it. refresh_token_id: uuid.UUID | None = None created_at: datetime = Field( - default_factory=lambda: datetime.now(timezone.utc), nullable=False + default_factory=lambda: datetime.now(timezone.utc), + nullable=False, + sa_type=UTCDateTime, ) @@ -163,10 +169,12 @@ class OAuthRefreshToken(SQLModel, table=True): #: Every token rotated out of one authorization shares this, so reusing an #: old one can revoke the whole line rather than just itself. family_id: uuid.UUID - expires_at: datetime + expires_at: datetime = Field(sa_type=UTCDateTime) revoked: bool = False created_at: datetime = Field( - default_factory=lambda: datetime.now(timezone.utc), nullable=False + default_factory=lambda: datetime.now(timezone.utc), + nullable=False, + sa_type=UTCDateTime, ) @@ -231,7 +239,7 @@ class MetricBucket(SQLModel, table=True): bucket: datetime = Field( primary_key=True, index=True, - sa_type=DateTime(timezone=True), # type: ignore + sa_type=UTCDateTime, ) executions: int = 0 errors: int = 0 @@ -255,7 +263,7 @@ class EngineEvent(SQLModel, table=True): id: int | None = Field(default=None, primary_key=True) ts: datetime = Field( index=True, - sa_type=DateTime(timezone=True), # type: ignore + sa_type=UTCDateTime, ) #: node_error, flow_quarantined, engine_degraded, …, or audit. type: str = Field(max_length=32, index=True) @@ -279,11 +287,11 @@ class FlowRun(SQLModel, table=True): source: str = "" started_at: datetime = Field( index=True, - sa_type=DateTime(timezone=True), # type: ignore + sa_type=UTCDateTime, ) finished_at: datetime | None = Field( default=None, - sa_type=DateTime(timezone=True), # type: ignore + sa_type=UTCDateTime, ) #: running, ok, error or abandoned. status: str = "running" @@ -337,15 +345,15 @@ class Run(SQLModel, table=True): labels: list[str] = Field(sa_column=Column(JSON), default_factory=list) created_at: datetime = Field( index=True, - sa_type=DateTime(timezone=True), # type: ignore + sa_type=UTCDateTime, ) started_at: datetime | None = Field( default=None, - sa_type=DateTime(timezone=True), # type: ignore + sa_type=UTCDateTime, ) finished_at: datetime | None = Field( default=None, - sa_type=DateTime(timezone=True), # type: ignore + sa_type=UTCDateTime, ) duration_ms: float = 0.0 #: The flow's declared outputs once it finished. @@ -355,7 +363,7 @@ class Run(SQLModel, table=True): engine: str = Field(default="", max_length=64) lease_at: datetime | None = Field( default=None, - sa_type=DateTime(timezone=True), # type: ignore + sa_type=UTCDateTime, ) actor: str = Field(default="", max_length=255) @@ -372,7 +380,7 @@ class RunNode(SQLModel, table=True): attempt: int = 1 started_at: datetime | None = Field( default=None, - sa_type=DateTime(timezone=True), # type: ignore + sa_type=UTCDateTime, ) duration_ms: float = 0.0 #: Which worker ran it: "local", or a remote worker's name. @@ -417,5 +425,5 @@ class RunArtifact(SQLModel, table=True): media_type: str = Field(default="application/octet-stream", max_length=128) created_at: datetime = Field( default_factory=lambda: datetime.now(timezone.utc), - sa_type=DateTime(timezone=True), # type: ignore + sa_type=UTCDateTime, ) diff --git a/backend/fluksio/tests_pre_start.py b/backend/fluksio/tests_pre_start.py deleted file mode 100644 index 5994154..0000000 --- a/backend/fluksio/tests_pre_start.py +++ /dev/null @@ -1,39 +0,0 @@ -import logging - -from sqlalchemy import Engine -from sqlmodel import Session, select -from tenacity import after_log, before_log, retry, stop_after_attempt, wait_fixed - -from fluksio.core.db import engine - -logging.basicConfig(level=logging.INFO) -logger = logging.getLogger(__name__) - -max_tries = 60 * 5 # 5 minutes -wait_seconds = 1 - - -@retry( - stop=stop_after_attempt(max_tries), - wait=wait_fixed(wait_seconds), - before=before_log(logger, logging.INFO), - after=after_log(logger, logging.WARN), -) -def init(db_engine: Engine) -> None: - try: - # Try to create session to check if DB is awake - with Session(db_engine) as session: - session.exec(select(1)) - except Exception as e: - logger.error(e) - raise e - - -def main() -> None: - logger.info("Initializing service") - init(engine) - logger.info("Service finished initializing") - - -if __name__ == "__main__": - main() diff --git a/backend/pyproject.toml b/backend/pyproject.toml index db06bec..4a5ad88 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -15,7 +15,6 @@ dependencies = [ "jinja2<4.0.0,>=3.1.4", "alembic<2.0.0,>=1.12.1", "httpx<1.0.0,>=0.25.1", - "psycopg[binary]<4.0.0,>=3.1.13", "sqlmodel<1.0.0,>=0.0.21", "pydantic-settings<3.0.0,>=2.2.1", "sentry-sdk[fastapi]<2.0.0,>=1.40.6", diff --git a/backend/scripts/prestart.sh b/backend/scripts/prestart.sh deleted file mode 100644 index 75770d6..0000000 --- a/backend/scripts/prestart.sh +++ /dev/null @@ -1,13 +0,0 @@ -#! /usr/bin/env bash - -set -e -set -x - -# Let the DB start -python -m fluksio.backend_pre_start - -# Run migrations -alembic upgrade head - -# Create initial data in DB -python -m fluksio.initial_data diff --git a/backend/scripts/soak.py b/backend/scripts/soak.py index 45acc65..2ca3bef 100644 --- a/backend/scripts/soak.py +++ b/backend/scripts/soak.py @@ -53,10 +53,10 @@ COMPOSE_PROJECT = "fluksio-app" REDIS = "fluksio-redis" API = "fluksio-api" BROKER = "fluksio-app-mosquitto-1" -DB = "fluksio-db" + #: The only containers this harness may ever name. The first three are the #: chaos targets; the database is read and cleaned up, never stopped. -ALLOWED = frozenset({REDIS, API, BROKER, DB}) +ALLOWED = frozenset({REDIS, API, BROKER}) class HostSafety(RuntimeError): @@ -805,25 +805,22 @@ def clear_redis() -> None: def clear_observability() -> None: """The collector wrote rollups, runs and events for the soak flows.""" - names = "', '".join(FLOWS) - statement = "; ".join( - f"DELETE FROM {table} WHERE flow IN ('{names}')" - for table in ("metric_minute", "flow_run", "engine_event") + names = ", ".join(repr(flow) for flow in FLOWS) + script = "\n".join( + [ + "import sqlite3", + "db = sqlite3.connect('/data/fluksio.db')", + *( + f"db.execute('DELETE FROM {table} WHERE flow IN ({names})')" + for table in ("metric_minute", "flow_run", "engine_event") + ), + "db.commit()", + ] ) try: - docker( - DB, - "exec", - DB, - "psql", - "-q", - "-U", - settings.POSTGRES_USER, - "-d", - settings.POSTGRES_DB, - "-c", - statement, - ) + # Through the api container: the database is a file on its volume, not + # a server of its own. + docker(API, "exec", API, "python", "-c", script) log.info("deleted observability rows for the soak flows") except Exception as exc: log.error("observability rows may be left behind: %s", exc) diff --git a/backend/scripts/tests-start.sh b/backend/scripts/tests-start.sh deleted file mode 100644 index 55f241a..0000000 --- a/backend/scripts/tests-start.sh +++ /dev/null @@ -1,7 +0,0 @@ -#! /usr/bin/env bash -set -e -set -x - -python -m fluksio.tests_pre_start - -bash scripts/test.sh "$@" diff --git a/backend/tests/__init__.py b/backend/tests/__init__.py index d3d75c7..dce203b 100644 --- a/backend/tests/__init__.py +++ b/backend/tests/__init__.py @@ -1,14 +1,32 @@ """Redirect the suite at its own database. -`app.core.db` builds the engine at import time and several modules bind that -object, so the name has to be in the environment before anything imports the -settings. Overriding it here — the first module pytest imports for the +`fluksio.core.db` builds the engine at import time and several modules bind +that object, so the URL has to be in the environment before anything imports +the settings. Overriding it here — the first module pytest imports for the package — keeps a test run from touching the development data. + +A file rather than `:memory:`: the suite starts the engine's own threads (the +run service, the metrics collector, the portal connector) against this engine, +and an in-memory database is per connection unless every one of them shares a +single connection, which those threads would then contend for. """ import os +import tempfile -os.environ["POSTGRES_DB"] = "app_test" +_DB_DIR = tempfile.mkdtemp(prefix="fluksio-test-") +os.environ.setdefault("DATABASE_URL", f"sqlite:///{_DB_DIR}/app_test.db") +# Nothing outside this directory: the suite must not pick up a checkout's .env +# and write into the development data. +os.environ.setdefault("FLUKSIO_ENV_FILE", os.path.join(_DB_DIR, "env")) +os.environ.setdefault("DATA_DIR", _DB_DIR) +# Named here rather than taken from a developer's .env, so the suite is the +# same run everywhere and a checkout's own superuser is never involved. +os.environ.setdefault("FIRST_SUPERUSER", "admin@example.com") +os.environ.setdefault("FIRST_SUPERUSER_PASSWORD", "testpassword") +# What `emails_enabled` needs; the tests that send mail patch SMTP_HOST on top +# of it. Nothing is delivered — the sender is faked where it matters. +os.environ.setdefault("EMAILS_FROM_EMAIL", "noreply@example.com") # The MCP session manager can only be entered once per instance, and the suite # builds a TestClient — and so a lifespan — per test module. Tests that want the # endpoint mount it themselves. diff --git a/backend/tests/api/routes/test_oauth.py b/backend/tests/api/routes/test_oauth.py index 10d6f3c..c6dcdaf 100644 --- a/backend/tests/api/routes/test_oauth.py +++ b/backend/tests/api/routes/test_oauth.py @@ -3,12 +3,15 @@ import base64 import hashlib import secrets +import uuid from urllib.parse import parse_qs, urlparse import pytest from fastapi.testclient import TestClient +from sqlmodel import Session, select from fluksio.core.config import settings +from fluksio.models import OAuthRefreshToken PREFIX = f"{settings.API_V1_STR}/oauth" REDIRECT = "http://127.0.0.1:41234/callback" @@ -235,6 +238,7 @@ def test_everything_is_refused_while_mcp_is_off( def test_one_agent_can_be_revoked_without_touching_the_others( client: TestClient, + db: Session, superuser_token_headers: dict[str, str], normal_user_token_headers: dict[str, str], ) -> None: @@ -272,3 +276,13 @@ def test_one_agent_can_be_revoked_without_touching_the_others( data={"grant_type": "refresh_token", "refresh_token": tokens["refresh_token"]}, ) assert refreshed.status_code == 400 + + # Went, rather than merely stopped working: nothing here deletes those rows + # itself, so an orphan is a foreign key the database is not enforcing — + # which SQLite does not do unless it is asked (`PRAGMA foreign_keys`). + left = db.exec( + select(OAuthRefreshToken).where( + OAuthRefreshToken.client_id == uuid.UUID(client_id) + ) + ).all() + assert left == [] diff --git a/backend/tests/api/routes/test_private.py b/backend/tests/api/routes/test_private.py index 179dee9..3cf22b2 100644 --- a/backend/tests/api/routes/test_private.py +++ b/backend/tests/api/routes/test_private.py @@ -1,3 +1,5 @@ +import uuid + import pytest from fastapi.testclient import TestClient from sqlmodel import Session, select @@ -20,7 +22,8 @@ def test_create_user(client: TestClient, db: Session) -> None: data = r.json() - user = db.exec(select(User).where(User.id == data["id"])).first() + # The id comes back from JSON as a string, and the column is a UUID. + user = db.exec(select(User).where(User.id == uuid.UUID(data["id"]))).first() assert user assert user.email == "pollo@listo.com" diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 0ee26cf..3dcfce5 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -1,18 +1,18 @@ import uuid from collections.abc import Generator from datetime import UTC, datetime +from pathlib import Path from typing import Any import pytest from cryptography.hazmat.primitives.asymmetric import rsa from fastapi.testclient import TestClient -from sqlalchemy import create_engine, text from sqlalchemy.engine import make_url -from sqlmodel import Session, SQLModel, select +from sqlmodel import Session, select from fluksio.cloud import config as cloud_config from fluksio.core.config import settings -from fluksio.core.db import engine, init_db +from fluksio.core.db import engine, prepare from fluksio.main import app from fluksio.models import User from tests.utils.portal import INSTALLATION_ID, ISSUER, jwks @@ -33,28 +33,25 @@ def flow_data(tmp_path_factory: pytest.TempPathFactory) -> Generator[None, None, @pytest.fixture(scope="session", autouse=True) def db() -> Generator[Session, None, None]: """Create the throwaway database `tests/__init__.py` points at, drop it after.""" - url = make_url(str(settings.SQLALCHEMY_DATABASE_URI)) - # The teardown drops this database, so refuse to run against anything but + url = make_url(settings.SQLALCHEMY_DATABASE_URI) + # The teardown deletes this database, so refuse to run against anything but # the dedicated test one. - assert url.database and url.database.endswith("_test"), url.database + assert url.get_backend_name() == "sqlite", url.get_backend_name() + assert url.database and url.database.endswith("_test.db"), url.database - maintenance = create_engine( - url.set(database="postgres"), isolation_level="AUTOCOMMIT" - ) - drop = text(f'DROP DATABASE IF EXISTS "{url.database}" WITH (FORCE)') - with maintenance.connect() as connection: - connection.execute(drop) - connection.execute(text(f'CREATE DATABASE "{url.database}"')) - - SQLModel.metadata.create_all(engine) + path = Path(url.database) + path.unlink(missing_ok=True) + # The real path, so the suite runs against a database built the way a + # deployment's is — including the alembic stamp the app's own startup + # would otherwise trip over. + prepare(engine) with Session(engine) as session: - init_db(session) yield session engine.dispose() - with maintenance.connect() as connection: - connection.execute(drop) - maintenance.dispose() + # The write-ahead log and its index are part of the database. + for suffix in ("", "-wal", "-shm"): + path.with_name(path.name + suffix).unlink(missing_ok=True) @pytest.fixture(scope="module") diff --git a/backend/tests/flow/test_connector.py b/backend/tests/flow/test_connector.py index af385ca..a2473ed 100644 --- a/backend/tests/flow/test_connector.py +++ b/backend/tests/flow/test_connector.py @@ -146,7 +146,9 @@ def test_a_connector_is_discovered_from_its_entry_point(monkeypatch): def load(self): return Sensor - monkeypatch.setattr("fluksio.flow.plugins.entry_points", lambda group: [FakeEntry()]) + monkeypatch.setattr( + "fluksio.flow.plugins.entry_points", lambda group: [FakeEntry()] + ) try: assert load_plugins() == ["test_sensor"] assert NODE_TYPES["test_sensor"].plugin == "fluksio-connector-test 0.1.0" @@ -166,7 +168,9 @@ def test_a_connector_written_for_another_contract_is_refused(monkeypatch): def load(self): return Outdated - monkeypatch.setattr("fluksio.flow.plugins.entry_points", lambda group: [FakeEntry()]) + monkeypatch.setattr( + "fluksio.flow.plugins.entry_points", lambda group: [FakeEntry()] + ) assert load_plugins() == [] assert "outdated" not in NODE_TYPES @@ -179,6 +183,8 @@ def test_a_connector_may_not_take_over_a_built_in_type(monkeypatch): def load(self): # pragma: no cover - never reached raise AssertionError("should not be loaded") - monkeypatch.setattr("fluksio.flow.plugins.entry_points", lambda group: [FakeEntry()]) + monkeypatch.setattr( + "fluksio.flow.plugins.entry_points", lambda group: [FakeEntry()] + ) assert load_plugins() == [] assert NODE_TYPES["mqtt"].plugin is None diff --git a/backend/tests/flow/test_workers.py b/backend/tests/flow/test_workers.py index e877f20..4a62fcc 100644 --- a/backend/tests/flow/test_workers.py +++ b/backend/tests/flow/test_workers.py @@ -123,7 +123,7 @@ def test_a_node_imports_the_standard_library_not_the_engines_own_modules(pool): def test_the_engines_secrets_are_not_in_a_workers_environment(pool, monkeypatch): monkeypatch.setenv("SECRET_KEY", "not-for-nodes") - monkeypatch.setenv("POSTGRES_PASSWORD", "not-for-nodes") + monkeypatch.setenv("DATABASE_URL", "not-for-nodes") monkeypatch.setenv("FLUKSIO_HARMLESS", "fine") # A fresh process, so it is built from the environment set just now. pool.respawn_all() @@ -132,7 +132,7 @@ def test_the_engines_secrets_are_not_in_a_workers_environment(pool, monkeypatch) pool, "import os\n\n\n" "def process():\n" - " return {'out': [k for k in ('SECRET_KEY', 'POSTGRES_PASSWORD',\n" + " return {'out': [k for k in ('SECRET_KEY', 'DATABASE_URL',\n" " 'FLUKSIO_HARMLESS') if k in os.environ]}\n", ) assert result == {"out": ["FLUKSIO_HARMLESS"]} diff --git a/backend/tests/scripts/__init__.py b/backend/tests/scripts/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/backend/tests/scripts/test_backend_pre_start.py b/backend/tests/scripts/test_backend_pre_start.py deleted file mode 100644 index 8c55be5..0000000 --- a/backend/tests/scripts/test_backend_pre_start.py +++ /dev/null @@ -1,33 +0,0 @@ -from unittest.mock import MagicMock, patch - -from sqlmodel import select - -from fluksio.backend_pre_start import init, logger - - -def test_init_successful_connection() -> None: - engine_mock = MagicMock() - - session_mock = MagicMock() - session_mock.__enter__.return_value = session_mock - - select1 = select(1) - - with ( - patch("fluksio.backend_pre_start.Session", return_value=session_mock), - patch("fluksio.backend_pre_start.select", return_value=select1), - patch.object(logger, "info"), - patch.object(logger, "error"), - patch.object(logger, "warn"), - ): - try: - init(engine_mock) - connection_successful = True - except Exception: - connection_successful = False - - assert connection_successful, ( - "The database connection should be successful and not raise an exception." - ) - - session_mock.exec.assert_called_once_with(select1) diff --git a/backend/tests/scripts/test_test_pre_start.py b/backend/tests/scripts/test_test_pre_start.py deleted file mode 100644 index a86f3ce..0000000 --- a/backend/tests/scripts/test_test_pre_start.py +++ /dev/null @@ -1,33 +0,0 @@ -from unittest.mock import MagicMock, patch - -from sqlmodel import select - -from fluksio.tests_pre_start import init, logger - - -def test_init_successful_connection() -> None: - engine_mock = MagicMock() - - session_mock = MagicMock() - session_mock.__enter__.return_value = session_mock - - select1 = select(1) - - with ( - patch("fluksio.tests_pre_start.Session", return_value=session_mock), - patch("fluksio.tests_pre_start.select", return_value=select1), - patch.object(logger, "info"), - patch.object(logger, "error"), - patch.object(logger, "warn"), - ): - try: - init(engine_mock) - connection_successful = True - except Exception: - connection_successful = False - - assert connection_successful, ( - "The database connection should be successful and not raise an exception." - ) - - session_mock.exec.assert_called_once_with(select1) diff --git a/backend/tests/test_cloud.py b/backend/tests/test_cloud.py index b4645f4..de5391d 100644 --- a/backend/tests/test_cloud.py +++ b/backend/tests/test_cloud.py @@ -206,7 +206,9 @@ def test_adding_a_remote_user_maps_and_revokes( return_value={"user_id": "portal-user-9", "email": "remote@example.com"} ), ) - with patch("fluksio.api.routes.cloud.httpx.post", return_value=portal_reply) as post: + with patch( + "fluksio.api.routes.cloud.httpx.post", return_value=portal_reply + ) as post: added = client.post( f"{settings.API_V1_STR}/cloud/users", headers=superuser_token_headers, diff --git a/backend/tests/test_db.py b/backend/tests/test_db.py new file mode 100644 index 0000000..183207f --- /dev/null +++ b/backend/tests/test_db.py @@ -0,0 +1,33 @@ +"""What the database has to keep true whatever dialect is under it.""" + +from datetime import datetime, timedelta, timezone + +from sqlmodel import Session, select + +from fluksio.core.db import engine +from fluksio.models import EngineEvent + +# The `db` fixture is session-scoped and autouse, so the schema is already there. + + +def test_a_stored_instant_comes_back_aware_and_in_utc() -> None: + """SQLite stores no offset, so the column type has to put one back. + + Without it a timestamp read from the database is naive: comparing it to + `datetime.now(timezone.utc)` raises, and serialising it hands the frontend + a time with no zone, which it reads as local. + """ + stamp = datetime(2026, 8, 21, 15, 30, tzinfo=timezone(timedelta(hours=2))) + with Session(engine) as session: + session.add( + EngineEvent(ts=stamp, type="node_error", flow="tz", node="n", detail="") + ) + session.commit() + + with Session(engine) as session: + stored = session.exec(select(EngineEvent).where(EngineEvent.flow == "tz")).one() + assert stored.ts.tzinfo is not None + assert stored.ts.utcoffset() == timedelta(0) + assert stored.ts == stamp + # And it still compares against an aware "now" rather than raising. + assert stored.ts < datetime.now(timezone.utc) diff --git a/development.md b/development.md index 3cff438..ec31314 100644 --- a/development.md +++ b/development.md @@ -19,12 +19,14 @@ default only (`export DOMAIN ?= …`), so `make dev DOMAIN=…` overrides it and reaches the recipes. **App only.** `cd app && make dev` starts this stack on its own Traefik with all host ports -published, in the foreground. `make dev-utils` starts just db, adminer, proxy, mailcatcher -and prestart — the useful half when the backend runs on the host. +published, in the foreground. `make dev-utils` starts just proxy and mailcatcher — the +useful half when the backend runs on the host. **No Docker.** `make install` once, then `make dev-backend` (FastAPI on :8000) and -`make dev-frontend` (Vite on :5173) in two terminals. A local stack's CORS list already -contains `http://localhost:5173`, so a host Vite server can talk to a containerised API. +`make dev-frontend` (Vite on :5173) in two terminals. The backend needs nothing running: its +database is SQLite in `DATA_DIR` (`backend/flow-data` by default), created on first start. +A local stack's CORS list already contains `http://localhost:5173`, so a host Vite server +can talk to a containerised API. Nothing invokes `docker compose` bare, and neither should you: both stacks keep their compose files in a `docker/` directory, so without `-p fluksio-app` they collide in a @@ -48,9 +50,9 @@ already being taken would stop the whole stack from starting. | http://api.localhost/docs | OpenAPI / Swagger UI | | http://localhost | marketing site (the `index` stack) | -Adminer and mailcatcher have no route here, so use the containers directly: -`docker exec -it fluksio-db psql -U postgres -d app` for the database, -`docker logs -f fluksio-app-mailcatcher-1` for captured mail. +Mailcatcher has no route here, so use the container directly: +`docker logs -f fluksio-app-mailcatcher-1` for captured mail. The database is a file on the +data volume: `docker exec -it fluksio-api sqlite3 /data/fluksio.db`. Standalone (`cd app && make dev`) publishes the full set: @@ -60,9 +62,7 @@ Standalone (`cd app && make dev`) publishes the full set: | http://localhost:5173 | SPA (the nginx image, not the Vite dev server) | | http://localhost:8000 | backend API | | http://localhost:8090 | Traefik dashboard | -| http://127.0.0.1:8080 | adminer | | http://localhost:1080 | mailcatcher web UI (SMTP on 1025) | -| localhost:5432 | Postgres | ## Hot reload, and what it misses @@ -94,35 +94,39 @@ port, on purpose — nothing outside the stack needs them. - **mailcatcher** — SMTP on 1025, web UI on 1080; the backend is pointed at it, so no development mail leaves the machine. -`adminer` and `redis` come from `compose.yml` and exist in production too; dev only differs -in that adminer gets a host port. Without `REDIS_HOST` the flow engine keeps state in -memory instead of Redis. +`redis` comes from `compose.yml` and exists in production too. Without `REDIS_HOST` the flow +engine keeps state in memory instead of Redis. `db` and `adminer` are still there but only +start under the `analytics` profile — they are Umami's Postgres now, not the engine's. ## Configuration and secrets The root `.env` is the source of truth. `scripts/setup.sh` creates `app/.env` from -`app/.env.example` and keeps the shared keys — domain, secret key, database password, -superuser — in step with it. Only the `.env.example` files are tracked; `make secrets` +`app/.env.example` and keeps the shared keys — domain, secret key, superuser — in step with +it. Only the `.env.example` files are tracked; `make secrets` reprints the generated values. A release that adds a key needs `scripts/setup.sh --secrets` before `make update`, so the key exists before the stack is rebuilt. -The backend settings load `../.env` relative to `backend/` through pydantic-settings, which -is why an exported environment variable always outranks the file — how CI points the suite -at a Docker-assigned Postgres port. +The backend settings load `../.env` relative to `backend/` through pydantic-settings — +`FLUKSIO_ENV_FILE` points that elsewhere, which is what an installed `fluksio` does. An +exported environment variable always outranks the file, which is how the suite pins its own +configuration. + +`DATA_DIR` is where this installation keeps everything: the SQLite database, the flow git +repository, `secrets.enc`, `oauth-key.pem`, `cloud.json`, artifacts and the user venv. Each +path derives from it and can still be set on its own — the compose stack spells all of them +out against `/data`. ## Tests `make test` is `test-backend` plus `test-frontend`. -**Backend.** `cd backend && uv run bash scripts/tests-start.sh` waits for Postgres, then -runs pytest under coverage; the HTML report lands in `backend/htmlcov`. Postgres is the only -service needed — mail is patched out in `tests/`, and the flow engine falls back to -`MemoryState` while `REDIS_HOST` is empty. `tests/__init__.py` pins the suite's own -environment before anything imports the settings: database `app_test` (created and dropped -per session, so a run never touches development data), plus `ENVIRONMENT=local` and -`DOMAIN=localhost`, so a checkout configured for a deployment cannot drag that deployment's -configuration into the run. `make test-backend` resolves `POSTGRES_SERVER` from the running -`fluksio-db` container's address, because the integrated stack publishes no 5432. +**Backend.** `cd backend && uv run bash scripts/test.sh` runs pytest under coverage; the +HTML report lands in `backend/htmlcov`. Nothing has to be running — mail is patched out in +`tests/`, and the flow engine falls back to `MemoryState` while `REDIS_HOST` is empty. +`tests/__init__.py` pins the suite's own environment before anything imports the settings: a +SQLite file in a temp directory (deleted per session, so a run never touches development +data), its own superuser, and `ENVIRONMENT=local` with `DOMAIN=localhost`, so a checkout +configured for a deployment cannot drag that deployment's configuration into the run. **Frontend (e2e).** `make test-frontend` runs the suite inside the pinned `mcr.microsoft.com/playwright:v-noble` image (version read from @@ -168,7 +172,7 @@ contract between the two frontends, and that target is the only thing checking i ## CI `.gitea/workflows/` runs four jobs, each writing `.env` from `.env.example` first: -`test-backend.yml` (pytest against a compose Postgres), `playwright.yml` (the e2e suite in +`test-backend.yml` (pytest, plus a check that both wheels build), `playwright.yml` (the e2e suite in two shards, one compose project each, the browser loading the SPA from the nginx `frontend` service), `pre-commit.yml` (the hooks above, plus strict mypy, which they do not cover) and `test-compose.yml` (a smoke test that the production images actually come up). diff --git a/docker/compose.ci.yml b/docker/compose.ci.yml index 18958ae..d0497de 100644 --- a/docker/compose.ci.yml +++ b/docker/compose.ci.yml @@ -18,11 +18,6 @@ services: db: container_name: !reset null - # The one service a job may need from the host: test-backend.yml runs pytest - # on the runner. Host side left to Docker, so concurrent jobs get a port - # each — `docker compose port db 5432` says which. - ports: !override - - "5432" redis: container_name: !reset null @@ -34,9 +29,6 @@ services: mailcatcher: ports: !reset [] - prestart: - container_name: !reset null - backend: container_name: !reset null ports: !reset [] diff --git a/docker/compose.dev.yml b/docker/compose.dev.yml index 8c3bc3e..b493208 100644 --- a/docker/compose.dev.yml +++ b/docker/compose.dev.yml @@ -35,6 +35,8 @@ services: - proxy - default + # Both only start under the analytics profile now — the engine's own state + # is SQLite on the data volume. db: restart: "no" ports: diff --git a/docker/compose.local.yml b/docker/compose.local.yml index 08d0c53..91f31cc 100644 --- a/docker/compose.local.yml +++ b/docker/compose.local.yml @@ -33,11 +33,6 @@ services: # talks to this backend rather than running one of its own. - BACKEND_CORS_ORIGINS=http://${DOMAIN:-localhost},http://app.${DOMAIN:-localhost},http://localhost:5173 - prestart: - environment: - - ENVIRONMENT=local - - FRONTEND_HOST=http://app.${DOMAIN:-localhost} - frontend: ports: !reset [] build: diff --git a/docker/compose.yml b/docker/compose.yml index 05d71b6..9aeaf35 100644 --- a/docker/compose.yml +++ b/docker/compose.yml @@ -7,14 +7,17 @@ services: + # Umami's database, and nothing else's. The engine keeps its own state in + # SQLite on the flow-data volume, which is why this is behind a profile now. db: image: postgres:18 container_name: fluksio-db restart: always + profiles: ["analytics"] security_opt: - no-new-privileges:true healthcheck: - test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER} -d ${POSTGRES_DB}"] + test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER:-postgres} -d ${POSTGRES_DB:-app}"] interval: 10s retries: 5 start_period: 30s @@ -25,9 +28,11 @@ services: - ../.env environment: - PGDATA=/var/lib/postgresql/data/pgdata - - POSTGRES_PASSWORD=${POSTGRES_PASSWORD?Variable not set} - - POSTGRES_USER=${POSTGRES_USER?Variable not set} - - POSTGRES_DB=${POSTGRES_DB?Variable not set} + # Not required of the stack at large any more: only the analytics + # profile starts this, and the engine has no Postgres to configure. + - POSTGRES_PASSWORD=${POSTGRES_PASSWORD:-postgres} + - POSTGRES_USER=${POSTGRES_USER:-postgres} + - POSTGRES_DB=${POSTGRES_DB:-app} redis: image: redis:8.2-alpine @@ -54,6 +59,7 @@ services: image: adminer container_name: fluksio-adminer restart: always + profiles: ["analytics"] security_opt: - no-new-privileges:true # Deliberately not on `proxy`: the Postgres UI is reachable from the host @@ -65,43 +71,6 @@ services: environment: - ADMINER_DESIGN=pepa-linha-dark - prestart: - image: '${DOCKER_IMAGE_BACKEND?Variable not set}:${TAG-latest}' - container_name: fluksio-prestart - build: - context: .. - dockerfile: backend/Dockerfile - security_opt: - - no-new-privileges:true - networks: - - proxy - - default - depends_on: - db: - condition: service_healthy - restart: true - command: bash scripts/prestart.sh - env_file: - - ../.env - environment: - - DOMAIN=${DOMAIN} - - FRONTEND_HOST=${FRONTEND_HOST?Variable not set} - - ENVIRONMENT=${ENVIRONMENT} - - BACKEND_CORS_ORIGINS=${BACKEND_CORS_ORIGINS} - - SECRET_KEY=${SECRET_KEY?Variable not set} - - FIRST_SUPERUSER=${FIRST_SUPERUSER?Variable not set} - - FIRST_SUPERUSER_PASSWORD=${FIRST_SUPERUSER_PASSWORD?Variable not set} - - SMTP_HOST=${SMTP_HOST} - - SMTP_USER=${SMTP_USER} - - SMTP_PASSWORD=${SMTP_PASSWORD} - - EMAILS_FROM_EMAIL=${EMAILS_FROM_EMAIL} - - POSTGRES_SERVER=db - - POSTGRES_PORT=${POSTGRES_PORT} - - POSTGRES_DB=${POSTGRES_DB} - - POSTGRES_USER=${POSTGRES_USER?Variable not set} - - POSTGRES_PASSWORD=${POSTGRES_PASSWORD?Variable not set} - - SENTRY_DSN=${SENTRY_DSN} - backend: image: '${DOCKER_IMAGE_BACKEND?Variable not set}:${TAG-latest}' container_name: fluksio-api @@ -112,14 +81,9 @@ services: - proxy - default depends_on: - db: - condition: service_healthy - restart: true redis: condition: service_healthy restart: true - prestart: - condition: service_completed_successfully env_file: - ../.env environment: @@ -134,14 +98,13 @@ services: - SMTP_USER=${SMTP_USER} - SMTP_PASSWORD=${SMTP_PASSWORD} - EMAILS_FROM_EMAIL=${EMAILS_FROM_EMAIL} - - POSTGRES_SERVER=db - - POSTGRES_PORT=${POSTGRES_PORT} - - POSTGRES_DB=${POSTGRES_DB} - - POSTGRES_USER=${POSTGRES_USER?Variable not set} - - POSTGRES_PASSWORD=${POSTGRES_PASSWORD?Variable not set} - SENTRY_DSN=${SENTRY_DSN} # Flow state survives a restart in Redis; without a host it stays in memory. - REDIS_HOST=redis + # Everything this installation keeps — the database, the flows, secrets, + # artifacts and the user venv — on one volume. The paths below derive + # from it; they are spelled out because each is load bearing. + - DATA_DIR=/data # Flows are files in a git repository; secrets sit encrypted beside it. - FLOWS_DIR=/data/flows - SECRETS_FILE=/data/secrets.enc