junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-06-30 13:10:47 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

UploadedFilePostgresRepository returns latest uploaded file by deal ID and type 🟩

Browse source 
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Daniel Roth 2026-06-09 11:54:14 +00:00
parent 53f0da8666
commit f08a75e103
1 changed files with 11 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
repositories/uploaded_file/uploaded_file_postgres_repository.py
View file

@ -2,7 +2,8 @@ from __future__ import annotations
from typing import Optional from typing import Optional
from sqlmodel import Session from sqlalchemy import select
from sqlmodel import Session, col
from infrastructure.postgres.uploaded_file_table import FileTypeEnum, UploadedFile from infrastructure.postgres.uploaded_file_table import FileTypeEnum, UploadedFile
@ -14,7 +15,14 @@ class UploadedFilePostgresRepository:
def get_latest_by_hubspot_deal_id( def get_latest_by_hubspot_deal_id(
self, hubspot_deal_id: str, file_type: FileTypeEnum self, hubspot_deal_id: str, file_type: FileTypeEnum
) -> Optional[UploadedFile]: ) -> Optional[UploadedFile]:
raise NotImplementedError stmt = (
select(UploadedFile)
.where(col(UploadedFile.hubspot_deal_id) == hubspot_deal_id)
.where(col(UploadedFile.file_type) == file_type.value)
.order_by(col(UploadedFile.s3_upload_timestamp).desc())
.limit(1)
)
return self._session.execute(stmt).scalars().one_or_none() # pyright: ignore[reportDeprecated]
def insert(self, uploaded_file: UploadedFile) -> None: def insert(self, uploaded_file: UploadedFile) -> None:
raise NotImplementedError self._session.add(uploaded_file)