Repository

// A interface mora no DOMINIO e fala a linguagem dele.
// Nada de Prisma, SQL, tabela ou coluna aqui.
export interface PedidosRepositorio {
  porId(id: PedidoId): Promise<Pedido | null>;
  pendentesHaMaisDe(horas: number, agora: Date): Promise<Pedido[]>;
  salvar(pedido: Pedido): Promise<void>;
}

// A implementacao mora na INFRA e e a unica que conhece o banco.
export class PedidosNoPostgres implements PedidosRepositorio {
  constructor(private db: PrismaClient) {}

  async pendentesHaMaisDe(horas: number, agora: Date): Promise<Pedido[]> {
    const linhas = await this.db.pedido.findMany({
      where: {
        status: { in: ["AGUARDANDO", "EM_ANALISE"] },
        criadoEm: { lt: new Date(agora.getTime() - horas * 3600_000) },
      },
    });
    // traduz linha -> objeto de dominio; o ORM para aqui
    return linhas.map(paraDominio);
  }

  async salvar(pedido: Pedido): Promise<void> {
    const dados = paraLinha(pedido);
    await this.db.pedido.upsert({
      where: { id: dados.id }, create: dados, update: dados,
    });
  }

  async porId(id: PedidoId) {
    const l = await this.db.pedido.findUnique({ where: { id: String(id) } });
    return l ? paraDominio(l) : null;
  }
}

// No teste, uma implementacao em memoria — sem banco, sem container.
export class PedidosEmMemoria implements PedidosRepositorio { /* ... */ }

from abc import ABC, abstractmethod
from datetime import datetime, timedelta

class PedidosRepositorio(ABC):
    """Contrato no dominio, na linguagem do negocio."""

    @abstractmethod
    def por_id(self, id: PedidoId) -> Pedido | None: ...

    @abstractmethod
    def pendentes_ha_mais_de(self, horas: int, agora: datetime) -> list[Pedido]: ...

    @abstractmethod
    def salvar(self, pedido: Pedido) -> None: ...


class PedidosNoPostgres(PedidosRepositorio):
    """Unica classe que sabe que existe SQLAlchemy."""

    def __init__(self, sessao):
        self._sessao = sessao

    def pendentes_ha_mais_de(self, horas: int, agora: datetime) -> list[Pedido]:
        corte = agora - timedelta(hours=horas)
        linhas = (
            self._sessao.query(PedidoORM)
            .filter(PedidoORM.status.in_(["AGUARDANDO", "EM_ANALISE"]))
            .filter(PedidoORM.criado_em < corte)
            .all()
        )
        return [para_dominio(l) for l in linhas]  # o ORM para aqui