This commit is contained in:
2026-07-17 17:59:43 +03:00
parent 507f4db7fd
commit 4512eb6e5c
4 changed files with 109 additions and 8 deletions

View File

@@ -9,6 +9,7 @@ from typing import Sequence, Union
from alembic import op from alembic import op
import sqlalchemy as sa import sqlalchemy as sa
from sqlalchemy.sql import text
# revision identifiers, used by Alembic. # revision identifiers, used by Alembic.
@@ -38,6 +39,7 @@ def upgrade() -> None:
) )
with op.batch_alter_table('permissions', schema=None) as batch_op: with op.batch_alter_table('permissions', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_permissions_id'), ['id'], unique=False) batch_op.create_index(batch_op.f('ix_permissions_id'), ['id'], unique=False)
op.execute(text("INSERT INTO permissions (permission) VALUES ('admin');"))
op.create_table('users', op.create_table('users',
sa.Column('id', sa.Uuid(), nullable=False), sa.Column('id', sa.Uuid(), nullable=False),
@@ -76,6 +78,7 @@ def upgrade() -> None:
sa.ForeignKeyConstraint(['user_id'], ['users.id'], name=op.f('fk_user_group_user_id_users')), sa.ForeignKeyConstraint(['user_id'], ['users.id'], name=op.f('fk_user_group_user_id_users')),
sa.PrimaryKeyConstraint('user_id', 'permission_group_id', name=op.f('pk_user_group')) sa.PrimaryKeyConstraint('user_id', 'permission_group_id', name=op.f('pk_user_group'))
) )
# ### end Alembic commands ### # ### end Alembic commands ###

View File

@@ -0,0 +1,54 @@
"""empty message
Revision ID: 9b30685906fb
Revises: 83b4c2ae88d3
Create Date: 2026-07-17 17:40:07.269967
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '9b30685906fb'
down_revision: Union[str, Sequence[str], None] = '83b4c2ae88d3'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('refresh_tokens',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('user_id', sa.Uuid(), nullable=False),
sa.Column('token_hash', sa.String(length=255), nullable=False),
sa.Column('device_info', sa.String(length=255), nullable=False),
sa.Column('ip_address', sa.String(length=45), nullable=False),
sa.Column('is_revoked', sa.Boolean(), nullable=False),
sa.Column('expires_at', sa.TIMESTAMP(), nullable=False),
sa.Column('created_at', sa.TIMESTAMP(), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.Column('replaced_by', sa.Integer(), nullable=True),
sa.ForeignKeyConstraint(['replaced_by'], ['refresh_tokens.id'], name=op.f('fk_refresh_tokens_replaced_by_refresh_tokens')),
sa.ForeignKeyConstraint(['user_id'], ['users.id'], name=op.f('fk_refresh_tokens_user_id_users'), ondelete='CASCADE'),
sa.PrimaryKeyConstraint('id', name=op.f('pk_refresh_tokens')),
sa.UniqueConstraint('token_hash', name=op.f('uq_refresh_tokens_token_hash'))
)
with op.batch_alter_table('refresh_tokens', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_refresh_tokens_id'), ['id'], unique=False)
batch_op.create_index(batch_op.f('ix_refresh_tokens_user_id'), ['user_id'], unique=False)
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
with op.batch_alter_table('refresh_tokens', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_refresh_tokens_user_id'))
batch_op.drop_index(batch_op.f('ix_refresh_tokens_id'))
op.drop_table('refresh_tokens')
# ### end Alembic commands ###

View File

@@ -1,6 +1,7 @@
from sqlalchemy import Table, create_engine, String, Boolean, MetaData, Column, ForeignKey from sqlalchemy import TIMESTAMP, Table, create_engine, String, Boolean, MetaData, Column, ForeignKey, func
from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase, relationship from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase, relationship
from uuid import UUID, uuid4 from uuid import UUID, uuid4
from datetime import datetime
engine = create_engine("sqlite:///DB/database.db", echo=True) engine = create_engine("sqlite:///DB/database.db", echo=True)
'''remember as a boilerplate, or just cp/pst''' '''remember as a boilerplate, or just cp/pst'''
@@ -27,6 +28,8 @@ class User(Model):
group:Mapped[list['PermissionsGroups']]=relationship(secondary="user_group", back_populates="user", lazy="selectin") group:Mapped[list['PermissionsGroups']]=relationship(secondary="user_group", back_populates="user", lazy="selectin")
direct_permissions:Mapped[list['Permissions']]=relationship(secondary="user_direct_permissions", back_populates="users_direct") direct_permissions:Mapped[list['Permissions']]=relationship(secondary="user_direct_permissions", back_populates="users_direct")
refresh_token:Mapped[list['RefreshTokens']]=relationship(back_populates="user")
def __repr__(self) -> str: def __repr__(self) -> str:
return f"ID: {self.id}, Name: {self.first_name}, Status: {self.status}" return f"ID: {self.id}, Name: {self.first_name}, Status: {self.status}"
@@ -73,4 +76,25 @@ group_permission = Table(
Model.metadata, Model.metadata,
Column("group_id", ForeignKey("groups_of_permissions.id"), primary_key=True), Column("group_id", ForeignKey("groups_of_permissions.id"), primary_key=True),
Column("permission_id", ForeignKey("permissions.id"), primary_key=True), Column("permission_id", ForeignKey("permissions.id"), primary_key=True),
) )
class RefreshTokens(Model):
__tablename__= "refresh_tokens"
id:Mapped[int]=mapped_column(primary_key=True, index=True)
user_id:Mapped[UUID]=mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)
token_hash:Mapped[str]=mapped_column(String(255), unique=True)
device_info:Mapped[str]=mapped_column(String(255))
ip_address:Mapped[str]=mapped_column(String(45))
is_revoked:Mapped[bool]=mapped_column(Boolean, default=False)
expires_at:Mapped[datetime]=mapped_column(TIMESTAMP)
created_at:Mapped[datetime]=mapped_column(TIMESTAMP, server_default=func.now())
replaced_by:Mapped[int|None]=mapped_column(ForeignKey("refresh_tokens.id"), nullable=True, default=None)
user:Mapped["User"]=relationship(back_populates="refresh_token")
def __repr__(self)->str:
return f'ID: {self.id}, USER_ID: {self.user_id}, Is_revoked: {self.is_revoked}, Token_Hash: {self.token_hash}'

View File

@@ -1,10 +1,12 @@
from dataclasses import replace
from datetime import datetime
from typing import Annotated from typing import Annotated
from pydantic import BaseModel, EmailStr, Field from pydantic import BaseModel, EmailStr, Field
from uuid import UUID from uuid import UUID
class Base(BaseModel): class Base(BaseModel):
pass model_config = {"from_attributes": True}
class UserCreate(Base): class UserCreate(Base):
@@ -28,7 +30,6 @@ class UserOut(Base):
permissions:Annotated[list[str], Field(..., description="permissions of the user")] permissions:Annotated[list[str], Field(..., description="permissions of the user")]
permission_groups:Annotated[list[str], Field(..., description="permissions groups of the user")] permission_groups:Annotated[list[str], Field(..., description="permissions groups of the user")]
model_config = {"from_attributes": True}
class UserUpdate(Base): class UserUpdate(Base):
@@ -42,11 +43,30 @@ class UserUpdate(Base):
class Permissions(Base): class Permissions(Base):
id:int id:int
permission_name:Annotated[str, Field(..., max_length=30, description="permission name")] permission:Annotated[str, Field(..., max_length=30, description="permission name")]
model_config = {"from_attributes": True}
class PermissionsOut(Base): class PermissionsOut(Base):
permission_name:Annotated[str, Field(..., max_length=30, description="permission name")] permission:Annotated[str, Field(..., max_length=30, description="permission name")]
model_config = {"from_attributes": True}
class PermissionsGroups(Base):
id:int
group:Annotated[str, Field(..., max_length=255, description="group name for the permissions")]
class PermissionsGroupsOut(Base):
group:Annotated[str, Field(..., max_length=255, description="group name for the permissions")]
class RefreshTokens(Base):
id:int
user_id:Annotated[UUID, Field(..., description="foreign key for the user")]
token_hash:Annotated[str, Field(..., description="token hash")]
device_info:Annotated[str, Field(..., description="User device info")]
ip_address:Annotated[str, Field(..., description="ip v4/v6 of the user")]
is_revoked:Annotated[bool|None, Field(None, description="revoke token if logout was made")]
expires_at:Annotated[datetime, Field(..., description="when token is going to be expired")]