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 f8abd91fc0
commit 73eeec29b1
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()