Repo

Models
Repo

Esta é a camada responsável pelos dados da aplicação, essa camada abstrai a fonte dos dados. O UseCase não sabe de onde veiram os dados, exemplo: banco, arquivo, consulta dados so sistema operacional e por isso o repo detem essa função.

No FwFlex a maioria dos dados estão em banco. No repo não escrevemos SQL puro, usamos o ORM SQLAlchemy.

Models

Para poder inserir no banco os dados, precisamos converter estes dados transformados em um objeto Python, para uma classe SQLBase que nos possibilita salvar no banco através do ORM.

class SQLUser(SQLBase):
    __tablename__ = "auth_users"

    id = sa.Column(sa.Integer, primary_key=True)
    username = sa.Column(
        sa.String(30), nullable=False, index=True, unique=True
    )
    fullname = sa.Column(sa.Unicode(64), nullable=True)
    email = sa.Column(sa.String(254), nullable=True, index=True, unique=True)
    password_hash = sa.Column(sa.String(150), nullable=True)
    superuser = sa.Column(sa.Boolean, nullable=False)
    has_console = sa.Column(sa.Boolean, nullable=False)
    allow_metrics = sa.Column(sa.Boolean, nullable=False)
    level = sa.Column(sa.String(3))
    enabled = sa.Column(sa.Boolean)
    created_at = sa.Column(sa.DateTime, nullable=True)
    updated_at = sa.Column(sa.DateTime, nullable=True)
    roles = relationship(SQLRole, secondary="auth_users_roles")

Repo

Para o Repo desenvolvemos o SQLBase. Que implementa a maioria dos métodos que utilizamos. Segue a baixo exemplo:

from typing import Set

from sqlalchemy.orm import Session, sessionmaker as SessionMaker

from itflex.common.datetime import as_naive_dt, from_naive_dt
from itflex.common.response import FieldError
from itflex.repo import SQLRepoBase
from itflex_auth.entities import User
from itflex_auth.models import SQLUser


class SQLUsersRepo(SQLRepoBase):
    def __init__(self, sessionmaker: SessionMaker):
        super().__init__(sessionmaker, SQLUser, user_to_db, db_to_user)

    def update_definition(self, db_user: SQLUser):
        return {
            SQLUser.name: db_user.name,
            SQLUser.description: db_user.description,
            SQLUser.updated_at: db_user.updated_at,
        }

    def find_duplicated_fields(
        self, session: Session, user: User
    ) -> Set[FieldError]:
        duplicate_fields = set()
        duplicate_usernames = None

        query = session.query(SQLUser)
        if user.id:
            query = query.filter(SQLUser.id != user.id)

        duplicate_usernames = query.filter_by(username=user.username).count()
        if duplicate_usernames:
            duplicate_fields.add("username")

        return duplicate_fields


# Transforma a entidade para objeto do banco
def user_to_db(user: User) -> SQLUser:
    return SQLUser(
        id=user.id,
        username=user.username,
        name=user.name,
        description=user.description,
        created_at=as_naive_dt(user.created_at),
        updated_at=as_naive_dt(user.updated_at),
    )


# Transforma o objeto do banco para entidade
def db_to_user(db_user: SQLUser) -> User:
    return User(
        id=db_user.id,
        username=db_to_user.username,
        name=db_user.name,
        description=db_user.description,
        created_at=from_naive_dt(db_user.created_at),
        updated_at=from_naive_dt(db_user.updated_at),
    )