wutta/wuttjamaican
3
0
Fork 0
Code Activity

feat: convert all uuid fields from str to proper UUID

Browse source 
ugh had to rewrite alembic migrations instead of just adding a new
one..  will be good to be past this hiccup
This commit is contained in:
Lance Edgar 2024-12-07 23:45:47 -06:00
parent 6a471b87c2
commit e1785ccfcc
8 changed files with 72 additions and 109 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

70
src/wuttjamaican/auth.py
View file

@ -26,6 +26,8 @@ Auth Handler
This defines the default :term:`auth handler`.
"""
import uuid as _uuid
from wuttjamaican.app import GenericHandler
@ -126,17 +128,28 @@ class AuthHandler(GenericHandler):
if not key:
return
# try to match on Role.uuid
role = session.get(model.Role, key)
if role:
return role
# maybe it is a uuid
if isinstance(key, _uuid.UUID):
role = session.get(model.Role, key)
if role:
return role
# try to match on Role.name
role = session.query(model.Role)\
.filter_by(name=key)\
.first()
if role:
return role
else: # assuming it is a string
# try to match on Role.uuid
try:
role = session.get(model.Role, _uuid.UUID(key))
if role:
return role
except ValueError:
pass
# try to match on Role.name
role = session.query(model.Role)\
.filter_by(name=key)\
.first()
if role:
return role
# try settings; if value then recurse
key = self.config.get(f'{self.appname}.role.{key}',
@ -178,21 +191,32 @@ class AuthHandler(GenericHandler):
if isinstance(obj, model.User):
return obj
# or maybe it is a string
# (nb. these lookups require a db session)
if isinstance(obj, str) and session:
# nb. these lookups require a db session
if session:
# try to match on User.uuid
user = session.get(model.User, obj)
if user:
return user
# or maybe it is a uuid
if isinstance(obj, _uuid.UUID):
user = session.get(model.User, obj)
if user:
return user
# try to match on User.username
user = session.query(model.User)\
.filter(model.User.username == obj)\
.first()
if user:
return user
# or maybe it is a string
elif isinstance(obj, str):
# try to match on User.uuid
try:
user = session.get(model.User, _uuid.UUID(obj))
if user:
return user
except ValueError:
pass
# try to match on User.username
user = session.query(model.User)\
.filter(model.User.username == obj)\
.first()
if user:
return user
# nb. obj is presumbly another type of object, e.g. Person

5
src/wuttjamaican/db/alembic/versions/3abcc44f7f91_add_people.py
View file

@ -9,6 +9,7 @@ from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
import wuttjamaican.db.util
# revision identifiers, used by Alembic.
@ -22,7 +23,7 @@ def upgrade() -> None:
# person
op.create_table('person',
sa.Column('uuid', sa.String(length=32), nullable=False),
sa.Column('uuid', wuttjamaican.db.util.UUID(), nullable=False),
sa.Column('full_name', sa.String(length=100), nullable=False),
sa.Column('first_name', sa.String(length=50), nullable=True),
sa.Column('middle_name', sa.String(length=50), nullable=True),
@ -31,7 +32,7 @@ def upgrade() -> None:
)
# user
op.add_column('user', sa.Column('person_uuid', sa.String(length=32), nullable=True))
op.add_column('user', sa.Column('person_uuid', wuttjamaican.db.util.UUID(), nullable=True))
op.create_foreign_key(op.f('fk_user_person_uuid_person'), 'user', 'person', ['person_uuid'], ['uuid'])

66
src/wuttjamaican/db/alembic/versions/6be0ed225f4d_convert_uuid_types.py
View file

@ -1,66 +0,0 @@
"""convert uuid types
Revision ID: 6be0ed225f4d
Revises: 6bf900765500
Create Date: 2024-11-30 17:03:08.930050
"""
import uuid as _uuid
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
import wuttjamaican.db.util
# revision identifiers, used by Alembic.
revision: str = '6be0ed225f4d'
down_revision: Union[str, None] = '6bf900765500'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# upgrade (convert uuid)
op.add_column('upgrade', sa.Column('uuid_true', wuttjamaican.db.util.UUID(), nullable=True))
upgrade = sa.sql.table('upgrade',
sa.sql.column('uuid'),
sa.sql.column('uuid_true'))
engine = op.get_bind()
cursor = engine.execute(upgrade.select())
for row in cursor.fetchall():
if row['uuid']:
uuid_true = _uuid.UUID(row['uuid'])
engine.execute(upgrade.update()\
.where(upgrade.c.uuid == row['uuid'])\
.values({'uuid_true': uuid_true}))
op.drop_constraint('pk_upgrade', 'upgrade', type_='primary')
op.drop_column('upgrade', 'uuid')
op.alter_column('upgrade', 'uuid_true', new_column_name='uuid')
op.create_primary_key('pk_upgrade', 'upgrade', ['uuid'])
def downgrade() -> None:
# upgrade (convert uuid)
op.add_column('upgrade', sa.Column('uuid_str', sa.VARCHAR(length=32), nullable=True))
upgrade = sa.sql.table('upgrade',
sa.sql.column('uuid'),
sa.sql.column('uuid_str'))
engine = op.get_bind()
cursor = engine.execute(upgrade.select())
for row in cursor.fetchall():
if row['uuid']:
uuid_str = row['uuid'].hex
engine.execute(upgrade.update()\
.where(upgrade.c.uuid == row['uuid'])\
.values({'uuid_str': uuid_str}))
op.drop_constraint('pk_upgrade', 'upgrade', type_='primary')
op.drop_column('upgrade', 'uuid')
op.alter_column('upgrade', 'uuid_str', new_column_name='uuid')
op.create_primary_key('pk_upgrade', 'upgrade', ['uuid'])

13
src/wuttjamaican/db/alembic/versions/d686f7abe3e0_add_users_roles.py
View file

@ -9,6 +9,7 @@ from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
import wuttjamaican.db.util
# revision identifiers, used by Alembic.
@ -22,7 +23,7 @@ def upgrade() -> None:
# role
op.create_table('role',
sa.Column('uuid', sa.String(length=32), nullable=False),
sa.Column('uuid', wuttjamaican.db.util.UUID(), nullable=False),
sa.Column('name', sa.String(length=100), nullable=False),
sa.Column('notes', sa.Text(), nullable=True),
sa.PrimaryKeyConstraint('uuid'),
@ -31,7 +32,7 @@ def upgrade() -> None:
# user
op.create_table('user',
sa.Column('uuid', sa.String(length=32), nullable=False),
sa.Column('uuid', wuttjamaican.db.util.UUID(), nullable=False),
sa.Column('username', sa.String(length=25), nullable=False),
sa.Column('password', sa.String(length=60), nullable=True),
sa.Column('active', sa.Boolean(), nullable=False),
@ -41,7 +42,7 @@ def upgrade() -> None:
# permission
op.create_table('permission',
sa.Column('role_uuid', sa.String(length=32), nullable=False),
sa.Column('role_uuid', wuttjamaican.db.util.UUID(), nullable=False),
sa.Column('permission', sa.String(length=254), nullable=False),
sa.ForeignKeyConstraint(['role_uuid'], ['role.uuid'], name=op.f('fk_permission_role_uuid_role')),
sa.PrimaryKeyConstraint('role_uuid', 'permission')
@ -49,9 +50,9 @@ def upgrade() -> None:
# user_x_role
op.create_table('user_x_role',
sa.Column('uuid', sa.String(length=32), nullable=False),
sa.Column('user_uuid', sa.String(length=32), nullable=False),
sa.Column('role_uuid', sa.String(length=32), nullable=False),
sa.Column('uuid', wuttjamaican.db.util.UUID(), nullable=False),
sa.Column('user_uuid', wuttjamaican.db.util.UUID(), nullable=False),
sa.Column('role_uuid', wuttjamaican.db.util.UUID(), nullable=False),
sa.ForeignKeyConstraint(['role_uuid'], ['role.uuid'], name=op.f('fk_user_x_role_role_uuid_role')),
sa.ForeignKeyConstraint(['user_uuid'], ['user.uuid'], name=op.f('fk_user_x_role_user_uuid_user')),
sa.PrimaryKeyConstraint('uuid')

7
src/wuttjamaican/db/alembic/versions/ebd75b9feaa7_add_upgrades.py
View file

@ -10,6 +10,7 @@ from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
import wuttjamaican.db.util
# revision identifiers, used by Alembic.
revision: str = 'ebd75b9feaa7'
@ -23,15 +24,15 @@ def upgrade() -> None:
# upgrade
sa.Enum('PENDING', 'EXECUTING', 'SUCCESS', 'FAILURE', name='upgradestatus').create(op.get_bind())
op.create_table('upgrade',
sa.Column('uuid', sa.String(length=32), nullable=False),
sa.Column('uuid', wuttjamaican.db.util.UUID(), nullable=False),
sa.Column('created', sa.DateTime(timezone=True), nullable=False),
sa.Column('created_by_uuid', sa.String(length=32), nullable=False),
sa.Column('created_by_uuid', wuttjamaican.db.util.UUID(), nullable=False),
sa.Column('description', sa.String(length=255), nullable=False),
sa.Column('notes', sa.Text(), nullable=True),
sa.Column('executing', sa.Boolean(), nullable=False),
sa.Column('status', postgresql.ENUM('PENDING', 'EXECUTING', 'SUCCESS', 'FAILURE', name='upgradestatus', create_type=False), nullable=False),
sa.Column('executed', sa.DateTime(timezone=True), nullable=True),
sa.Column('executed_by_uuid', sa.String(length=32), nullable=True),
sa.Column('executed_by_uuid', wuttjamaican.db.util.UUID(), nullable=True),
sa.Column('exit_code', sa.Integer(), nullable=True),
sa.ForeignKeyConstraint(['created_by_uuid'], ['user.uuid'], name=op.f('fk_upgrade_created_by_uuid_user')),
sa.ForeignKeyConstraint(['executed_by_uuid'], ['user.uuid'], name=op.f('fk_upgrade_executed_by_uuid_user')),

8
src/wuttjamaican/db/util.py
View file

@ -30,7 +30,7 @@ import sqlalchemy as sa
from sqlalchemy import orm
from sqlalchemy.dialects.postgresql import UUID as PGUUID
from wuttjamaican.util import make_uuid
from wuttjamaican.util import make_true_uuid
# nb. this convention comes from upstream docs
@ -111,10 +111,10 @@ def uuid_column(*args, **kwargs):
Returns a UUID column for use as a table's primary key.
"""
if not args:
args = (sa.String(length=32),)
args = (UUID(),)
kwargs.setdefault('primary_key', True)
kwargs.setdefault('nullable', False)
kwargs.setdefault('default', make_uuid)
kwargs.setdefault('default', make_true_uuid)
return sa.Column(*args, **kwargs)
@ -126,7 +126,7 @@ def uuid_fk_column(target_column, *args, **kwargs):
e.g. ``'user.uuid'``.
"""
if not args:
args = (sa.String(length=32), sa.ForeignKey(target_column))
args = (UUID(), sa.ForeignKey(target_column))
return sa.Column(*args, **kwargs)

6
tests/db/test_util.py
View file

@ -114,8 +114,7 @@ else:
def test_basic(self):
column = mod.uuid_column()
self.assertIsInstance(column, sa.Column)
self.assertIsInstance(column.type, sa.String)
self.assertEqual(column.type.length, 32)
self.assertIsInstance(column.type, mod.UUID)
class TestUUIDFKColumn(TestCase):
@ -123,8 +122,7 @@ else:
def test_basic(self):
column = mod.uuid_fk_column('foo.bar')
self.assertIsInstance(column, sa.Column)
self.assertIsInstance(column.type, sa.String)
self.assertEqual(column.type.length, 32)
self.assertIsInstance(column.type, mod.UUID)
class TestMakeTopoSortkey(DataTestCase):

6
tests/test_auth.py
View file

@ -93,7 +93,7 @@ else:
self.config.usedb = True
role = self.handler.get_role(self.session, 'mykey')
self.assertIsNone(role)
setting = model.Setting(name='wutta.role.mykey', value=myrole.uuid)
setting = model.Setting(name='wutta.role.mykey', value=myrole.uuid.hex)
self.session.add(setting)
self.session.commit()
role = self.handler.get_role(self.session, 'mykey')
@ -117,6 +117,10 @@ else:
user = self.handler.get_user(myuser.uuid, session=self.session)
self.assertIs(user, myuser)
# match on User.uuid (str)
user = self.handler.get_user(myuser.uuid.hex, session=self.session)
self.assertIs(user, myuser)
# match on User.username
user = self.handler.get_user(myuser.username, session=self.session)
self.assertIs(user, myuser)