122 lines
5.4 KiB
Python
122 lines
5.4 KiB
Python
from datetime import datetime
|
|
from uuid import UUID, uuid4
|
|
|
|
from sqlalchemy import (
|
|
TIMESTAMP,
|
|
Boolean,
|
|
Column,
|
|
ForeignKey,
|
|
MetaData,
|
|
String,
|
|
Table,
|
|
Uuid,
|
|
func,
|
|
)
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
|
|
|
|
from src.models.configs_read.env import env_settings
|
|
|
|
engine = create_async_engine(f"postgresql+asyncpg://{env_settings.DB_USER}:{env_settings.DB_PASSWORD}@{env_settings.DB_HOST}:{env_settings.DB_PORT}/{env_settings.DB_POSTGRESS}",
|
|
pool_size=20, # сколько соединений держать открытыми постоянно
|
|
max_overflow=10, # сколько доп. соединений можно создать при пиковой нагрузке
|
|
pool_timeout=30, # сколько ждать свободное соединение, прежде чем упасть с ошибкой
|
|
pool_pre_ping=True, # проверять соединение перед использованием (ловит "протухшие" соединения)
|
|
)
|
|
|
|
'''remember as a boilerplate, or just cp/pst'''
|
|
class Model(DeclarativeBase):
|
|
metadata = MetaData(naming_convention={
|
|
"ix": "ix_%(column_0_label)s",
|
|
"uq": "uq_%(table_name)s_%(column_0_name)s",
|
|
"ck": "ck_%(table_name)s_%(constraint_name)s",
|
|
"fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
|
|
"pk": "pk_%(table_name)s",
|
|
})
|
|
|
|
class User(Model):
|
|
__tablename__ = "users"
|
|
|
|
id:Mapped[UUID] = mapped_column(Uuid(as_uuid=True),default=uuid4,primary_key=True)
|
|
first_name:Mapped[str] = mapped_column(String(64), index=True)
|
|
last_name:Mapped[str]=mapped_column(String(64), index=True)
|
|
middle_name:Mapped[str]=mapped_column(String(64), index=True)
|
|
email:Mapped[str]=mapped_column(String(255), index=True, unique=True)
|
|
status:Mapped[bool]=mapped_column(Boolean, default=True)
|
|
hashed_password:Mapped[str]=mapped_column(String(255))
|
|
|
|
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", lazy="selectin")
|
|
|
|
refresh_token:Mapped[list['RefreshTokens']]=relationship(back_populates="user")
|
|
report:Mapped[list["Stored"]]=relationship(back_populates="user")
|
|
|
|
def __repr__(self) -> str:
|
|
return f"ID: {self.id}, Name: {self.first_name}, Status: {self.status}"
|
|
|
|
class PermissionsGroups(Model):
|
|
__tablename__="groups_of_permissions"
|
|
|
|
id:Mapped[int]=mapped_column(primary_key=True, index=True)
|
|
group:Mapped[str]=mapped_column(String(255), unique=True)
|
|
|
|
user:Mapped[list['User']]=relationship(secondary="user_group", back_populates="group")
|
|
permissions:Mapped[list["Permissions"]]=relationship(secondary="group_permission", back_populates="group", lazy="selectin")
|
|
|
|
def __repr__(self)->str:
|
|
return f"ID: {self.id}, Permissions: {self.group}"
|
|
|
|
class Permissions(Model):
|
|
__tablename__ = "permissions"
|
|
|
|
id:Mapped[int]=mapped_column(primary_key=True, index=True)
|
|
permission:Mapped[str]=mapped_column(String(255), unique=True)
|
|
|
|
group:Mapped[list['PermissionsGroups']]=relationship(secondary="group_permission", back_populates="permissions")
|
|
users_direct:Mapped[list["User"]]=relationship(secondary="user_direct_permissions", back_populates="direct_permissions")
|
|
|
|
def __repr__(self)->str:
|
|
return f"ID: {self.id}, Permissions: {self.permission}"
|
|
|
|
user_group_of_permissions=Table(
|
|
"user_group",
|
|
Model.metadata,
|
|
Column("user_id", ForeignKey("users.id", ondelete="CASCADE"), primary_key=True),
|
|
Column("permission_group_id", ForeignKey("groups_of_permissions.id"), primary_key=True)
|
|
)
|
|
|
|
user_permission=Table(
|
|
"user_direct_permissions",
|
|
Model.metadata,
|
|
Column("user_id",ForeignKey("users.id", ondelete="CASCADE"), primary_key=True),
|
|
Column("permission_id", ForeignKey("permissions.id"), primary_key=True)
|
|
)
|
|
|
|
group_permission = Table(
|
|
"group_permission",
|
|
Model.metadata,
|
|
Column("group_id", ForeignKey("groups_of_permissions.id"), primary_key=True),
|
|
Column("permission_id", ForeignKey("permissions.id"), primary_key=True),
|
|
)
|
|
|
|
|
|
class RefreshTokens(Model):
|
|
|
|
__tablename__= "refresh_tokens"
|
|
|
|
id:Mapped[UUID]=mapped_column(Uuid(as_uuid=True),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(timezone=True))
|
|
created_at:Mapped[datetime]=mapped_column(TIMESTAMP(timezone=True), server_default=func.now())
|
|
replaced_by:Mapped[UUID|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}' |