LeoVasanko/paskia
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

CRUD store and delete on the DB classes directly.

Browse Source
This commit is contained in:
Leo Vasanko
2026-02-10 22:19:02 +00:00
parent f0d360758c
commit 2f2934db6b
3 changed files with 109 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
paskia/db/bootstrap.py
+6 -6
View File
@@ -70,7 +70,7 @@ def bootstrap(
orgs={org_uuid: True}, # Grant to org
)
perm_admin.uuid = perm_admin_uuid
_ops._db.permissions[perm_admin_uuid] = perm_admin
perm_admin.store()
# Create auth:org:admin permission
perm_org_admin = Permission(
@@ -79,12 +79,12 @@ def bootstrap(
orgs={org_uuid: True}, # Grant to org
)
perm_org_admin.uuid = perm_org_admin_uuid
_ops._db.permissions[perm_org_admin_uuid] = perm_org_admin
perm_org_admin.store()
# Create organization
new_org = Org.create(display_name=org_name)
new_org.uuid = org_uuid
_ops._db.orgs[org_uuid] = new_org
new_org.store()
# Create Administration role with both permissions
admin_role = Role(
@@ -93,7 +93,7 @@ def bootstrap(
permissions={perm_admin_uuid: True, perm_org_admin_uuid: True},
)
admin_role.uuid = role_uuid
_ops._db.roles[role_uuid] = admin_role
admin_role.store()
# Create admin user
admin_user = User(
@@ -104,7 +104,7 @@ def bootstrap(
visits=0,
)
admin_user.uuid = user_uuid
_ops._db.users[user_uuid] = admin_user
admin_user.store()
# Create reset token
reset_token, reset_passphrase = ResetToken.create(
@@ -113,7 +113,7 @@ def bootstrap(
token_type="admin bootstrap",
passphrase=reset_passphrase,
)
_ops._db.reset_tokens[reset_token.key] = reset_token
reset_token.store()
# Set config if provided
if config is not None:
paskia/db/operations.py
+19 -45
View File
@@ -128,7 +128,7 @@ def create_permission(perm: Permission, *, ctx: SessionContext | None = None) ->
if perm.uuid in _db.permissions:
raise ValueError(f"Permission {perm.uuid} already exists")
with _db.transaction("admin:create_permission", ctx):
_db.permissions[perm.uuid] = perm
perm.store()
def update_permission(
@@ -156,10 +156,7 @@ def delete_permission(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
if uuid not in _db.permissions:
raise ValueError(f"Permission {uuid} not found")
with _db.transaction("admin:delete_permission", ctx):
# Remove this permission from all roles
for role in _db.roles.values():
role.permissions.pop(uuid, None)
del _db.permissions[uuid]
_db.permissions[uuid].delete()
def create_org(org: Org, *, ctx: SessionContext | None = None) -> None:
@@ -173,7 +170,7 @@ def create_org(org: Org, *, ctx: SessionContext | None = None) -> None:
with _db.transaction("admin:create_org", ctx):
new_org = Org.create(display_name=org.display_name, created_at=now)
new_org.uuid = org.uuid
_db.orgs[org.uuid] = new_org
new_org.store()
# Create Administration role with org admin permission
admin_role_uuid = uuid7.create(now)
@@ -190,7 +187,7 @@ def create_org(org: Org, *, ctx: SessionContext | None = None) -> None:
permissions=role_permissions,
)
admin_role.uuid = admin_role_uuid
_db.roles[admin_role_uuid] = admin_role
admin_role.store()
def update_org_name(
@@ -211,16 +208,7 @@ def delete_org(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
if uuid not in _db.orgs:
raise ValueError(f"Organization {uuid} not found")
with _db.transaction("admin:delete_org", ctx):
org = _db.orgs[uuid]
# Remove org from all permissions
for p in _db.permissions.values():
p.orgs.pop(uuid, None)
# Delete roles in this org and their users
for role in org.roles:
for user in role.users:
del _db.users[user.uuid]
del _db.roles[role.uuid]
del _db.orgs[uuid]
_db.orgs[uuid].delete()
def add_permission_to_org(
@@ -264,7 +252,7 @@ def create_role(role: Role, *, ctx: SessionContext | None = None) -> None:
if role.org_uuid not in _db.orgs:
raise ValueError(f"Organization {role.org_uuid} not found")
with _db.transaction("admin:create_role", ctx):
_db.roles[role.uuid] = role
role.store()
def update_role_name(
@@ -317,7 +305,7 @@ def delete_role(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
if role.users:
raise ValueError(f"Cannot delete role {uuid}: users still assigned")
with _db.transaction("admin:delete_role", ctx):
del _db.roles[uuid]
_db.roles[uuid].delete()
def create_user(new_user: User, *, ctx: SessionContext | None = None) -> None:
@@ -327,7 +315,7 @@ def create_user(new_user: User, *, ctx: SessionContext | None = None) -> None:
if new_user.role_uuid not in _db.roles:
raise ValueError(f"Role {new_user.role_uuid} not found")
with _db.transaction("admin:create_user", ctx):
_db.users[new_user.uuid] = new_user
new_user.store()
def update_user_display_name(
@@ -386,19 +374,8 @@ def delete_user(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
"""Delete user and their credentials/sessions."""
if uuid not in _db.users:
raise ValueError(f"User {uuid} not found")
user = _db.users[uuid]
with _db.transaction("admin:delete_user", ctx):
# Delete credentials
for cred in user.credentials:
del _db.credentials[cred.uuid]
# Delete sessions
for sess in user.sessions:
del _db.sessions[sess.key]
# Delete reset tokens (iterate over dict items to get correct keys)
for key, token in list(_db.reset_tokens.items()):
if token.user_uuid == uuid:
del _db.reset_tokens[key]
del _db.users[uuid]
_db.users[uuid].delete()
def create_credential(cred: Credential, *, ctx: SessionContext | None = None) -> None:
@@ -408,7 +385,7 @@ def create_credential(cred: Credential, *, ctx: SessionContext | None = None) ->
if cred.user_uuid not in _db.users:
raise ValueError(f"User {cred.user_uuid} not found")
with _db.transaction("create_credential", ctx):
_db.credentials[cred.uuid] = cred
cred.store()
def update_credential_sign_count(
@@ -444,11 +421,7 @@ def delete_credential(
if cred.user_uuid != user_uuid:
raise ValueError(f"Credential {uuid} does not belong to user {user_uuid}")
with _db.transaction("delete_credential", ctx):
# Delete all sessions using this credential
for sess in cred.sessions:
print(sess, repr(sess.key))
del _db.sessions[sess.key]
del _db.credentials[uuid]
cred.delete()
def create_session(
@@ -523,7 +496,7 @@ def delete_session(
if key not in _db.sessions:
raise ValueError("Session not found")
with _db.transaction(action, ctx):
del _db.sessions[key]
_db.sessions[key].delete()
def delete_sessions_for_user(
@@ -540,7 +513,7 @@ def delete_sessions_for_user(
return
with _db.transaction("admin:delete_sessions_for_user", ctx):
for sess in user.sessions:
del _db.sessions[sess.key]
sess.delete()
def create_reset_token(
@@ -570,7 +543,7 @@ def create_reset_token(
if token.key in _db.reset_tokens:
raise ValueError("Reset token already exists")
with _db.transaction("create_reset_token", ctx, user=user):
_db.reset_tokens[token.key] = token
token.store()
return passphrase
@@ -579,7 +552,7 @@ def delete_reset_token(key: bytes, *, ctx: SessionContext | None = None) -> None
if key not in _db.reset_tokens:
raise ValueError("Reset token not found")
with _db.transaction("delete_reset_token", ctx):
del _db.reset_tokens[key]
_db.reset_tokens[key].delete()
# -------------------------------------------------------------------------
@@ -697,15 +670,16 @@ def create_credential_session(
credential.last_verified = now
# Create credential
_db.credentials[credential.uuid] = credential
credential.store()
# Store session and record visit
session.store(now)
# Delete reset token if provided
if reset_key:
if reset_key in _db.reset_tokens:
del _db.reset_tokens[reset_key]
token = _db.reset_tokens.get(reset_key)
if token:
token.delete()
return session.key
paskia/db/structs.py
+84
View File
@@ -48,6 +48,20 @@ class Permission(msgspec.Struct, dict=True, omit_defaults=True):
if org_uuid in db.data().orgs
]
def store(self) -> None:
"""Store this permission in the database. Must be called inside a transaction."""
db.data().permissions[self.uuid] = self
def delete(self) -> None:
"""Delete this permission and remove it from all roles.
Must be called inside a transaction.
"""
_data = db.data()
for role in _data.roles.values():
role.permissions.pop(self.uuid, None)
del _data.permissions[self.uuid]
@classmethod
def create(
cls,
@@ -86,6 +100,24 @@ class Org(msgspec.Struct, dict=True):
"""Get all permissions that this organization can grant."""
return [p for p in db.data().permissions.values() if self.uuid in p.orgs]
def store(self) -> None:
"""Store this organization in the database. Must be called inside a transaction."""
db.data().orgs[self.uuid] = self
def delete(self) -> None:
"""Delete this org and cascade to roles, users. Remove from permissions.
Must be called inside a transaction.
"""
_data = db.data()
for p in _data.permissions.values():
p.orgs.pop(self.uuid, None)
for role in self.roles:
for user in role.users:
del _data.users[user.uuid]
del _data.roles[role.uuid]
del _data.orgs[self.uuid]
@classmethod
def create(cls, display_name: str, created_at: datetime | None = None) -> Org:
"""Create a new Org with auto-generated uuid7."""
@@ -135,6 +167,14 @@ class Role(msgspec.Struct, dict=True, omit_defaults=True):
"""Get all users that have this role."""
return [u for u in db.data().users.values() if u.role_uuid == self.uuid]
def store(self) -> None:
"""Store this role in the database. Must be called inside a transaction."""
db.data().roles[self.uuid] = self
def delete(self) -> None:
"""Delete this role from the database. Must be called inside a transaction."""
del db.data().roles[self.uuid]
@classmethod
def create(
cls,
@@ -199,6 +239,24 @@ class User(msgspec.Struct, dict=True, omit_defaults=True):
"""Get all reset tokens for this user."""
return [t for t in db.data().reset_tokens.values() if t.user_uuid == self.uuid]
def store(self) -> None:
"""Store this user in the database. Must be called inside a transaction."""
db.data().users[self.uuid] = self
def delete(self) -> None:
"""Delete this user and cascade to credentials, sessions, reset tokens.
Must be called inside a transaction.
"""
_data = db.data()
for cred in self.credentials:
del _data.credentials[cred.uuid]
for sess in self.sessions:
del _data.sessions[sess.key]
for token in self.reset_tokens:
del _data.reset_tokens[token.key]
del _data.users[self.uuid]
@classmethod
def create(
cls,
@@ -250,6 +308,20 @@ class Credential(msgspec.Struct, dict=True):
s for s in db.data().sessions.values() if s.credential_uuid == self.uuid
]
def store(self) -> None:
"""Store this credential in the database. Must be called inside a transaction."""
db.data().credentials[self.uuid] = self
def delete(self) -> None:
"""Delete this credential and all its sessions.
Must be called inside a transaction.
"""
_data = db.data()
for sess in self.sessions:
del _data.sessions[sess.key]
del _data.credentials[self.uuid]
@classmethod
def create(
cls,
@@ -325,6 +397,10 @@ class Session(msgspec.Struct, dict=True):
_data.users[self.user_uuid].last_seen = last_seen
_data.users[self.user_uuid].visits += 1
def delete(self) -> None:
"""Delete this session from the database. Must be called inside a transaction."""
del db.data().sessions[self.key]
@classmethod
def create(
cls,
@@ -372,6 +448,14 @@ class ResetToken(msgspec.Struct, dict=True):
"""Get the User object for this reset token."""
return db.data().users[self.user_uuid]
def store(self) -> None:
"""Store this reset token in the database. Must be called inside a transaction."""
db.data().reset_tokens[self.key] = self
def delete(self) -> None:
"""Delete this reset token from the database. Must be called inside a transaction."""
del db.data().reset_tokens[self.key]
@classmethod
def create(
cls,