Fluksio/app
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Keep the engine's state in SQLite, not Postgres

Browse Source 
One process owns this database — the image has run a single uvicorn
worker for that reason since the four-engines bug — so a file beside the
flows is the honest shape for it, and it is what lets `fluksio serve`
need no infrastructure at all. Live values, node execution and the work
queue never came here anyway; what does is a rollup a minute at a time,
a row per cascade and the run history, and WAL keeps the readers going
while that one writer works.

DATA_DIR is now the one setting that moves everything an installation
keeps; the rest derive from it and the images still spell theirs out.
The schema is prepared in-process at startup, so the prestart service is
gone, and the ten Postgres-only revisions collapse into one portable
baseline.

Three things only worked because psycopg was casting for us: a token's
subject arriving as a string where the column is a UUID, `greatest`, and
`date_bin`. The timestamps needed a column type of their own — SQLite
stores no offset, and a naive datetime read back either raises against an
aware `now` or serialises as local time.

Postgres stays in the stack only for Umami, behind the analytics profile.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
stroblme
2026-08-21 22:19:45 +02:00
co-authored by Claude Opus 5
parent 2c369ac75f
commit 961a8f881d
51 changed files with 841 additions and 1089 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/fluksio/alembic/env.py
+37 -7
View File
@@ -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():
backend/fluksio/alembic/versions/087c44e16304_add_observability_rollups_events_and_.py
-77
View File
@@ -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 ###
backend/fluksio/alembic/versions/1a31ce608336_add_cascade_delete_relationships.py
-37
View File
@@ -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 ###
backend/fluksio/alembic/versions/59e2606ce144_add_oauth_client_code_and_refresh_token.py
-70
View File
@@ -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 ###
backend/fluksio/alembic/versions/9c0a54914c78_add_max_length_for_string_varchar_.py
-69
View File
@@ -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)
backend/fluksio/alembic/versions/a3f1c07b52d9_add_run_tables.py
-126
View File
@@ -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')
backend/fluksio/alembic/versions/b7c41d2f8a30_drop_item_table.py
-40
View File
@@ -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"),
)
backend/fluksio/alembic/versions/c4e8b2170f93_add_portal_sub_to_user.py
-35
View File
@@ -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')
backend/fluksio/alembic/versions/d98dd8ec85a3_edit_replace_id_integers_in_all_models_.py
-90
View File
@@ -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'])
backend/fluksio/alembic/versions/e2412789c190_initialize_models.py
-54
View File
@@ -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 ###
backend/fluksio/alembic/versions/ee1b4b4426a3_baseline_schema.py
+337
View File
@@ -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")
backend/fluksio/alembic/versions/fe56fa70289e_add_created_at_to_user_and_item.py
-31
View File
@@ -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 ###
backend/fluksio/api/deps.py
+11 -1
View File
@@ -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(
backend/fluksio/api/routes/observability.py
+20 -20
View File
@@ -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
backend/fluksio/backend_pre_start.py
-39
View File
@@ -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()
backend/fluksio/cloud/connector.py
+2 -1
View File
@@ -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
backend/fluksio/core/config.py
+59 -22
View File
@@ -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()
backend/fluksio/core/db.py
+96 -14
View File
@@ -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)
backend/fluksio/core/types.py
+44
View File
@@ -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)
backend/fluksio/flow/metrics.py
+4 -4
View File
@@ -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
},
)
backend/fluksio/flow/runs.py
+2 -4
View File
@@ -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"],
backend/fluksio/flow/workers.py
+1 -1
View File
@@ -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")
backend/fluksio/initial_data.py
-23
View File
@@ -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()
backend/fluksio/main.py
+6
View File
@@ -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()
backend/fluksio/models.py
+26 -18
View File
@@ -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,
)
backend/fluksio/tests_pre_start.py
-39
View File
@@ -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()
backend/pyproject.toml
-1
View File
@@ -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",
backend/scripts/prestart.sh
-13
View File
@@ -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
backend/scripts/soak.py
+16 -19
View File
@@ -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)
backend/scripts/tests-start.sh
-7
View File
@@ -1,7 +0,0 @@
#! /usr/bin/env bash
set -e
set -x
python -m fluksio.tests_pre_start
bash scripts/test.sh "$@"
backend/tests/__init__.py
+22 -4
View File
@@ -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.
backend/tests/api/routes/test_oauth.py
+14
View File
@@ -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 == []
backend/tests/api/routes/test_private.py
+4 -1
View File
@@ -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"
backend/tests/conftest.py
+16 -19
View File
@@ -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")
backend/tests/flow/test_connector.py
+9 -3
View File
@@ -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
backend/tests/flow/test_workers.py
+2 -2
View File
@@ -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"]}
backend/tests/scripts/__init__.py
View File
backend/tests/scripts/test_backend_pre_start.py
-33
View File
@@ -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)
backend/tests/scripts/test_test_pre_start.py
-33
View File
@@ -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)
backend/tests/test_cloud.py
+3 -1
View File
@@ -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,
backend/tests/test_db.py
+33
View File
@@ -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)