[feat] add api_key table

create model for api_key
create migration for alembic
This commit is contained in:
Tim Gröger 2024-10-14 06:24:19 +02:00
parent 81080404fb
commit 2f7fdec492
4 changed files with 99 additions and 8 deletions

View File

@ -0,0 +1,41 @@
"""Add APIKeys
Revision ID: f9aa4cafa982
Revises: 20482a003db8
Create Date: 2024-10-11 13:04:21.877288
"""
import sqlalchemy as sa
from alembic import op
import flaschengeist
# revision identifiers, used by Alembic.
revision = "f9aa4cafa982"
down_revision = "20482a003db8"
branch_labels = ()
depends_on = None
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table(
"api_key",
sa.Column("expires", flaschengeist.database.types.UtcDateTime(), nullable=True),
sa.Column("token", sa.String(length=32), nullable=True),
sa.Column("lifetime", sa.Integer(), nullable=True),
sa.Column("id", flaschengeist.database.types.Serial(), nullable=False),
sa.Column("user_id", flaschengeist.database.types.Serial(), nullable=True),
sa.ForeignKeyConstraint(["user_id"], ["user.id"], name=op.f("fk_api_key_user_id_user")),
sa.PrimaryKeyConstraint("id", name=op.f("pk_api_key")),
sa.UniqueConstraint("token", name=op.f("uq_api_key_token")),
)
# ### end Alembic commands ###
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table("api_key")
# ### end Alembic commands ###

View File

@ -1,5 +1,6 @@
from .api_key import *
from .image import *
from .notification import *
from .plugin import *
from .session import * from .session import *
from .user import * from .user import *
from .plugin import *
from .notification import *
from .image import *

View File

@ -0,0 +1,48 @@
from __future__ import \
annotations # TODO: Remove if python requirement is >= 3.12 (? PEP 563 is defered)
from datetime import datetime, timedelta, timezone
from secrets import compare_digest
from .. import logger
from ..database import db
from ..database.types import ModelSerializeMixin, Serial, UtcDateTime
class ApiKey(db.Model, ModelSerializeMixin):
"""Model for a Session
Args:
expires: Is a Datetime from current Time.
user: Is an User.
token: String to verify access later.
"""
__allow_unmapped__ = True
__tablename__ = "api_key"
expires: datetime = db.Column(UtcDateTime, nullable=True)
token: str = db.Column(db.String(32), unique=True)
lifetime: int = db.Column(db.Integer, nullable=True)
userid: str = ""
_id = db.Column("id", Serial, primary_key=True)
_user_id = db.Column("user_id", Serial, db.ForeignKey("user.id"))
user_: User = db.relationship("User", back_populates="api_keys_")
@property
def userid(self):
return self.user_.userid
def refresh(self):
"""Update the Timestamp
Update the Timestamp to the current Time.
"""
logger.debug("update timestamp from session with token {{ {} }}".format(self.token))
self.expires = datetime.now(timezone.utc) + timedelta(seconds=self.lifetime)
def __eq__(self, token):
if isinstance(token, str):
return compare_digest(self.token, token)
else:
return super(Session, self).__eq__(token)

View File

@ -1,13 +1,13 @@
from __future__ import ( from __future__ import \
annotations, annotations # TODO: Remove if python requirement is >= 3.12 (? PEP 563 is defered)
) # TODO: Remove if python requirement is >= 3.12 (? PEP 563 is defered)
from typing import Optional, Union, List
from datetime import date, datetime from datetime import date, datetime
from typing import List, Optional, Union
from sqlalchemy.orm.collections import attribute_mapped_collection from sqlalchemy.orm.collections import attribute_mapped_collection
from ..database import db from ..database import db
from ..database.types import ModelSerializeMixin, UtcDateTime, Serial from ..database.types import ModelSerializeMixin, Serial, UtcDateTime
association_table = db.Table( association_table = db.Table(
"user_x_role", "user_x_role",
@ -71,6 +71,7 @@ class User(db.Model, ModelSerializeMixin):
id_ = db.Column("id", Serial, primary_key=True) id_ = db.Column("id", Serial, primary_key=True)
roles_: List[Role] = db.relationship("Role", secondary=association_table, cascade="save-update, merge") roles_: List[Role] = db.relationship("Role", secondary=association_table, cascade="save-update, merge")
sessions_: List[Session] = db.relationship("Session", back_populates="user_", cascade="all, delete, delete-orphan") sessions_: List[Session] = db.relationship("Session", back_populates="user_", cascade="all, delete, delete-orphan")
api_keys_: List[ApiKey] = db.relationship("ApiKey", back_populates="user_", cascade="all, delete, delete-orphan")
avatar_: Optional[Image] = db.relationship("Image", cascade="all, delete, delete-orphan", single_parent=True) avatar_: Optional[Image] = db.relationship("Image", cascade="all, delete, delete-orphan", single_parent=True)
reset_requests_: List["_PasswordReset"] = db.relationship("_PasswordReset", cascade="all, delete, delete-orphan") reset_requests_: List["_PasswordReset"] = db.relationship("_PasswordReset", cascade="all, delete, delete-orphan")