fix: format all code with black

and from now on should not deviate from that...
This commit is contained in:
Lance Edgar 2025-08-31 12:45:58 -05:00
parent e20ef31ff7
commit 1b01df79e2
9 changed files with 312 additions and 144 deletions

View file

@ -3,4 +3,4 @@
from importlib.metadata import version
__version__ = version('Wutta-Continuum')
__version__ = version("Wutta-Continuum")

View file

@ -40,5 +40,6 @@ class WuttaContinuumAppProvider(AppProvider):
This checks the config value as described in
:doc:`/narr/install`; default will be ``False``.
"""
return self.config.get_bool('wutta_continuum.enable_versioning',
usedb=False, default=False)
return self.config.get_bool(
"wutta_continuum.enable_versioning", usedb=False, default=False
)

View file

@ -42,19 +42,23 @@ class WuttaContinuumConfigExtension(WuttaConfigExtension):
This adds a startup hook, which can optionally turn on the
SQLAlchemy-Continuum versioning features for the main app DB.
"""
key = 'wutta_continuum'
key = "wutta_continuum"
def startup(self, config):
""" """
# only do this if config enables it
if not config.get_bool('wutta_continuum.enable_versioning',
usedb=False, default=False):
if not config.get_bool(
"wutta_continuum.enable_versioning", usedb=False, default=False
):
return
# create wutta plugin, to assign user and ip address
spec = config.get('wutta_continuum.wutta_plugin_spec',
usedb=False,
default='wutta_continuum.conf:WuttaContinuumPlugin')
spec = config.get(
"wutta_continuum.wutta_plugin_spec",
usedb=False,
default="wutta_continuum.conf:WuttaContinuumPlugin",
)
WuttaPlugin = load_object(spec)
# tell sqlalchemy-continuum to do its thing
@ -109,10 +113,10 @@ class WuttaContinuumPlugin(Plugin):
remote_addr = self.get_remote_addr(uow, session)
if remote_addr:
kwargs['remote_addr'] = remote_addr
kwargs["remote_addr"] = remote_addr
user_id = self.get_user_id(uow, session)
if user_id:
kwargs['user_id'] = user_id
kwargs["user_id"] = user_id
return kwargs

View file

@ -5,6 +5,7 @@ Revises: 71406251b8e7
Create Date: 2024-11-24 17:39:57.415425
"""
from typing import Sequence, Union
from alembic import op
@ -12,8 +13,8 @@ import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '0a5f8ac0cd06'
down_revision: Union[str, None] = '71406251b8e7'
revision: str = "0a5f8ac0cd06"
down_revision: Union[str, None] = "71406251b8e7"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
@ -21,10 +22,13 @@ depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# user
op.add_column('user_version', sa.Column('prevent_edit', sa.Boolean(), autoincrement=False, nullable=True))
op.add_column(
"user_version",
sa.Column("prevent_edit", sa.Boolean(), autoincrement=False, nullable=True),
)
def downgrade() -> None:
# user
op.drop_column('user_version', 'prevent_edit')
op.drop_column("user_version", "prevent_edit")

View file

@ -1,10 +1,11 @@
"""first versioning tables
Revision ID: 71406251b8e7
Revises:
Revises:
Create Date: 2024-08-27 18:28:31.488291
"""
from typing import Sequence, Union
from alembic import op
@ -13,131 +14,283 @@ import wuttjamaican.db.util
# revision identifiers, used by Alembic.
revision: str = '71406251b8e7'
revision: str = "71406251b8e7"
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = ('wutta_continuum',)
branch_labels: Union[str, Sequence[str], None] = ("wutta_continuum",)
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# transaction
op.create_table('transaction',
sa.Column('issued_at', sa.DateTime(), nullable=True),
sa.Column('id', sa.BigInteger(), autoincrement=True, nullable=False),
sa.Column('remote_addr', sa.String(length=50), nullable=True),
sa.Column('user_id', wuttjamaican.db.util.UUID(), nullable=True),
sa.ForeignKeyConstraint(['user_id'], ['user.uuid'], name=op.f('fk_transaction_user_id_user')),
sa.PrimaryKeyConstraint('id', name=op.f('pk_transaction'))
)
op.create_index(op.f('ix_transaction_user_id'), 'transaction', ['user_id'], unique=False)
op.create_table(
"transaction",
sa.Column("issued_at", sa.DateTime(), nullable=True),
sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
sa.Column("remote_addr", sa.String(length=50), nullable=True),
sa.Column("user_id", wuttjamaican.db.util.UUID(), nullable=True),
sa.ForeignKeyConstraint(
["user_id"], ["user.uuid"], name=op.f("fk_transaction_user_id_user")
),
sa.PrimaryKeyConstraint("id", name=op.f("pk_transaction")),
)
op.create_index(
op.f("ix_transaction_user_id"), "transaction", ["user_id"], unique=False
)
# person
op.create_table('person_version',
sa.Column('uuid', wuttjamaican.db.util.UUID(), autoincrement=False, nullable=False),
sa.Column('full_name', sa.String(length=100), autoincrement=False, nullable=True),
sa.Column('first_name', sa.String(length=50), autoincrement=False, nullable=True),
sa.Column('middle_name', sa.String(length=50), autoincrement=False, nullable=True),
sa.Column('last_name', sa.String(length=50), autoincrement=False, nullable=True),
sa.Column('transaction_id', sa.BigInteger(), autoincrement=False, nullable=False),
sa.Column('end_transaction_id', sa.BigInteger(), nullable=True),
sa.Column('operation_type', sa.SmallInteger(), nullable=False),
sa.PrimaryKeyConstraint('uuid', 'transaction_id', name=op.f('pk_person_version'))
)
op.create_index(op.f('ix_person_version_end_transaction_id'), 'person_version', ['end_transaction_id'], unique=False)
op.create_index(op.f('ix_person_version_operation_type'), 'person_version', ['operation_type'], unique=False)
op.create_index(op.f('ix_person_version_transaction_id'), 'person_version', ['transaction_id'], unique=False)
op.create_table(
"person_version",
sa.Column(
"uuid", wuttjamaican.db.util.UUID(), autoincrement=False, nullable=False
),
sa.Column(
"full_name", sa.String(length=100), autoincrement=False, nullable=True
),
sa.Column(
"first_name", sa.String(length=50), autoincrement=False, nullable=True
),
sa.Column(
"middle_name", sa.String(length=50), autoincrement=False, nullable=True
),
sa.Column(
"last_name", sa.String(length=50), autoincrement=False, nullable=True
),
sa.Column(
"transaction_id", sa.BigInteger(), autoincrement=False, nullable=False
),
sa.Column("end_transaction_id", sa.BigInteger(), nullable=True),
sa.Column("operation_type", sa.SmallInteger(), nullable=False),
sa.PrimaryKeyConstraint(
"uuid", "transaction_id", name=op.f("pk_person_version")
),
)
op.create_index(
op.f("ix_person_version_end_transaction_id"),
"person_version",
["end_transaction_id"],
unique=False,
)
op.create_index(
op.f("ix_person_version_operation_type"),
"person_version",
["operation_type"],
unique=False,
)
op.create_index(
op.f("ix_person_version_transaction_id"),
"person_version",
["transaction_id"],
unique=False,
)
# user
op.create_table('user_version',
sa.Column('uuid', wuttjamaican.db.util.UUID(), autoincrement=False, nullable=False),
sa.Column('username', sa.String(length=25), autoincrement=False, nullable=True),
sa.Column('password', sa.String(length=60), autoincrement=False, nullable=True),
sa.Column('person_uuid', wuttjamaican.db.util.UUID(), autoincrement=False, nullable=True),
sa.Column('active', sa.Boolean(), autoincrement=False, nullable=True),
sa.Column('transaction_id', sa.BigInteger(), autoincrement=False, nullable=False),
sa.Column('end_transaction_id', sa.BigInteger(), nullable=True),
sa.Column('operation_type', sa.SmallInteger(), nullable=False),
sa.PrimaryKeyConstraint('uuid', 'transaction_id', name=op.f('pk_user_version'))
)
op.create_index(op.f('ix_user_version_end_transaction_id'), 'user_version', ['end_transaction_id'], unique=False)
op.create_index(op.f('ix_user_version_operation_type'), 'user_version', ['operation_type'], unique=False)
op.create_index(op.f('ix_user_version_transaction_id'), 'user_version', ['transaction_id'], unique=False)
op.create_table(
"user_version",
sa.Column(
"uuid", wuttjamaican.db.util.UUID(), autoincrement=False, nullable=False
),
sa.Column("username", sa.String(length=25), autoincrement=False, nullable=True),
sa.Column("password", sa.String(length=60), autoincrement=False, nullable=True),
sa.Column(
"person_uuid",
wuttjamaican.db.util.UUID(),
autoincrement=False,
nullable=True,
),
sa.Column("active", sa.Boolean(), autoincrement=False, nullable=True),
sa.Column(
"transaction_id", sa.BigInteger(), autoincrement=False, nullable=False
),
sa.Column("end_transaction_id", sa.BigInteger(), nullable=True),
sa.Column("operation_type", sa.SmallInteger(), nullable=False),
sa.PrimaryKeyConstraint("uuid", "transaction_id", name=op.f("pk_user_version")),
)
op.create_index(
op.f("ix_user_version_end_transaction_id"),
"user_version",
["end_transaction_id"],
unique=False,
)
op.create_index(
op.f("ix_user_version_operation_type"),
"user_version",
["operation_type"],
unique=False,
)
op.create_index(
op.f("ix_user_version_transaction_id"),
"user_version",
["transaction_id"],
unique=False,
)
# role
op.create_table('role_version',
sa.Column('uuid', wuttjamaican.db.util.UUID(), autoincrement=False, nullable=False),
sa.Column('name', sa.String(length=100), autoincrement=False, nullable=True),
sa.Column('notes', sa.Text(), autoincrement=False, nullable=True),
sa.Column('transaction_id', sa.BigInteger(), autoincrement=False, nullable=False),
sa.Column('end_transaction_id', sa.BigInteger(), nullable=True),
sa.Column('operation_type', sa.SmallInteger(), nullable=False),
sa.PrimaryKeyConstraint('uuid', 'transaction_id', name=op.f('pk_role_version'))
)
op.create_index(op.f('ix_role_version_end_transaction_id'), 'role_version', ['end_transaction_id'], unique=False)
op.create_index(op.f('ix_role_version_operation_type'), 'role_version', ['operation_type'], unique=False)
op.create_index(op.f('ix_role_version_transaction_id'), 'role_version', ['transaction_id'], unique=False)
op.create_table(
"role_version",
sa.Column(
"uuid", wuttjamaican.db.util.UUID(), autoincrement=False, nullable=False
),
sa.Column("name", sa.String(length=100), autoincrement=False, nullable=True),
sa.Column("notes", sa.Text(), autoincrement=False, nullable=True),
sa.Column(
"transaction_id", sa.BigInteger(), autoincrement=False, nullable=False
),
sa.Column("end_transaction_id", sa.BigInteger(), nullable=True),
sa.Column("operation_type", sa.SmallInteger(), nullable=False),
sa.PrimaryKeyConstraint("uuid", "transaction_id", name=op.f("pk_role_version")),
)
op.create_index(
op.f("ix_role_version_end_transaction_id"),
"role_version",
["end_transaction_id"],
unique=False,
)
op.create_index(
op.f("ix_role_version_operation_type"),
"role_version",
["operation_type"],
unique=False,
)
op.create_index(
op.f("ix_role_version_transaction_id"),
"role_version",
["transaction_id"],
unique=False,
)
# user_x_role
op.create_table('user_x_role_version',
sa.Column('uuid', wuttjamaican.db.util.UUID(), autoincrement=False, nullable=False),
sa.Column('user_uuid', wuttjamaican.db.util.UUID(), autoincrement=False, nullable=True),
sa.Column('role_uuid', wuttjamaican.db.util.UUID(), autoincrement=False, nullable=True),
sa.Column('transaction_id', sa.BigInteger(), autoincrement=False, nullable=False),
sa.Column('end_transaction_id', sa.BigInteger(), nullable=True),
sa.Column('operation_type', sa.SmallInteger(), nullable=False),
sa.PrimaryKeyConstraint('uuid', 'transaction_id', name=op.f('pk_user_x_role_version'))
)
op.create_index(op.f('ix_user_x_role_version_end_transaction_id'), 'user_x_role_version', ['end_transaction_id'], unique=False)
op.create_index(op.f('ix_user_x_role_version_operation_type'), 'user_x_role_version', ['operation_type'], unique=False)
op.create_index(op.f('ix_user_x_role_version_transaction_id'), 'user_x_role_version', ['transaction_id'], unique=False)
op.create_table(
"user_x_role_version",
sa.Column(
"uuid", wuttjamaican.db.util.UUID(), autoincrement=False, nullable=False
),
sa.Column(
"user_uuid", wuttjamaican.db.util.UUID(), autoincrement=False, nullable=True
),
sa.Column(
"role_uuid", wuttjamaican.db.util.UUID(), autoincrement=False, nullable=True
),
sa.Column(
"transaction_id", sa.BigInteger(), autoincrement=False, nullable=False
),
sa.Column("end_transaction_id", sa.BigInteger(), nullable=True),
sa.Column("operation_type", sa.SmallInteger(), nullable=False),
sa.PrimaryKeyConstraint(
"uuid", "transaction_id", name=op.f("pk_user_x_role_version")
),
)
op.create_index(
op.f("ix_user_x_role_version_end_transaction_id"),
"user_x_role_version",
["end_transaction_id"],
unique=False,
)
op.create_index(
op.f("ix_user_x_role_version_operation_type"),
"user_x_role_version",
["operation_type"],
unique=False,
)
op.create_index(
op.f("ix_user_x_role_version_transaction_id"),
"user_x_role_version",
["transaction_id"],
unique=False,
)
# permission
op.create_table('permission_version',
sa.Column('role_uuid', wuttjamaican.db.util.UUID(), autoincrement=False, nullable=False),
sa.Column('permission', sa.String(length=254), autoincrement=False, nullable=False),
sa.Column('transaction_id', sa.BigInteger(), autoincrement=False, nullable=False),
sa.Column('end_transaction_id', sa.BigInteger(), nullable=True),
sa.Column('operation_type', sa.SmallInteger(), nullable=False),
sa.PrimaryKeyConstraint('role_uuid', 'permission', 'transaction_id', name=op.f('pk_permission_version'))
)
op.create_index(op.f('ix_permission_version_end_transaction_id'), 'permission_version', ['end_transaction_id'], unique=False)
op.create_index(op.f('ix_permission_version_operation_type'), 'permission_version', ['operation_type'], unique=False)
op.create_index(op.f('ix_permission_version_transaction_id'), 'permission_version', ['transaction_id'], unique=False)
op.create_table(
"permission_version",
sa.Column(
"role_uuid",
wuttjamaican.db.util.UUID(),
autoincrement=False,
nullable=False,
),
sa.Column(
"permission", sa.String(length=254), autoincrement=False, nullable=False
),
sa.Column(
"transaction_id", sa.BigInteger(), autoincrement=False, nullable=False
),
sa.Column("end_transaction_id", sa.BigInteger(), nullable=True),
sa.Column("operation_type", sa.SmallInteger(), nullable=False),
sa.PrimaryKeyConstraint(
"role_uuid",
"permission",
"transaction_id",
name=op.f("pk_permission_version"),
),
)
op.create_index(
op.f("ix_permission_version_end_transaction_id"),
"permission_version",
["end_transaction_id"],
unique=False,
)
op.create_index(
op.f("ix_permission_version_operation_type"),
"permission_version",
["operation_type"],
unique=False,
)
op.create_index(
op.f("ix_permission_version_transaction_id"),
"permission_version",
["transaction_id"],
unique=False,
)
def downgrade() -> None:
# permission
op.drop_index(op.f('ix_permission_version_transaction_id'), table_name='permission_version')
op.drop_index(op.f('ix_permission_version_operation_type'), table_name='permission_version')
op.drop_index(op.f('ix_permission_version_end_transaction_id'), table_name='permission_version')
op.drop_table('permission_version')
op.drop_index(
op.f("ix_permission_version_transaction_id"), table_name="permission_version"
)
op.drop_index(
op.f("ix_permission_version_operation_type"), table_name="permission_version"
)
op.drop_index(
op.f("ix_permission_version_end_transaction_id"),
table_name="permission_version",
)
op.drop_table("permission_version")
# user_x_role
op.drop_index(op.f('ix_user_x_role_version_transaction_id'), table_name='user_x_role_version')
op.drop_index(op.f('ix_user_x_role_version_operation_type'), table_name='user_x_role_version')
op.drop_index(op.f('ix_user_x_role_version_end_transaction_id'), table_name='user_x_role_version')
op.drop_table('user_x_role_version')
op.drop_index(
op.f("ix_user_x_role_version_transaction_id"), table_name="user_x_role_version"
)
op.drop_index(
op.f("ix_user_x_role_version_operation_type"), table_name="user_x_role_version"
)
op.drop_index(
op.f("ix_user_x_role_version_end_transaction_id"),
table_name="user_x_role_version",
)
op.drop_table("user_x_role_version")
# role
op.drop_index(op.f('ix_role_version_transaction_id'), table_name='role_version')
op.drop_index(op.f('ix_role_version_operation_type'), table_name='role_version')
op.drop_index(op.f('ix_role_version_end_transaction_id'), table_name='role_version')
op.drop_table('role_version')
op.drop_index(op.f("ix_role_version_transaction_id"), table_name="role_version")
op.drop_index(op.f("ix_role_version_operation_type"), table_name="role_version")
op.drop_index(op.f("ix_role_version_end_transaction_id"), table_name="role_version")
op.drop_table("role_version")
# user
op.drop_index(op.f('ix_user_version_transaction_id'), table_name='user_version')
op.drop_index(op.f('ix_user_version_operation_type'), table_name='user_version')
op.drop_index(op.f('ix_user_version_end_transaction_id'), table_name='user_version')
op.drop_table('user_version')
op.drop_index(op.f("ix_user_version_transaction_id"), table_name="user_version")
op.drop_index(op.f("ix_user_version_operation_type"), table_name="user_version")
op.drop_index(op.f("ix_user_version_end_transaction_id"), table_name="user_version")
op.drop_table("user_version")
# person
op.drop_index(op.f('ix_person_version_transaction_id'), table_name='person_version')
op.drop_index(op.f('ix_person_version_operation_type'), table_name='person_version')
op.drop_index(op.f('ix_person_version_end_transaction_id'), table_name='person_version')
op.drop_table('person_version')
op.drop_index(op.f("ix_person_version_transaction_id"), table_name="person_version")
op.drop_index(op.f("ix_person_version_operation_type"), table_name="person_version")
op.drop_index(
op.f("ix_person_version_end_transaction_id"), table_name="person_version"
)
op.drop_table("person_version")
# transaction
op.drop_index(op.f('ix_transaction_user_id'), table_name='transaction')
op.drop_table('transaction')
op.drop_index(op.f("ix_transaction_user_id"), table_name="transaction")
op.drop_table("transaction")