++

# Conflicts:
#	backend/bloom/infra/database/sql_model.py
#	backend/bloom/infra/repositories/repository_port.py
#	backend/bloom/routers/v1/ports.py
#	backend/bloom/routers/v1/vessels.py

.env.template

@@ -3,6 +3,7 @@
 ###############################################################################################
 # these values are used in the local docker env. You can use "localhost" hostname
 # if you run the application without docker 
+POSTGRES_DRIVER=postgresql
 POSTGRES_HOSTNAME=postgres_bloom
 POSTGRES_USER=bloom_user
 POSTGRES_PASSWORD=bloom

backend/bloom/config.py

@@ -41,6 +41,7 @@ class Settings(BaseSettings):
                                   default=5432)
 
     postgres_db:str = Field(min_length=1,max_length=32,pattern=r'^(?:[a-zA-Z]|_)[\w\d_]*$')
+    postgres_schema:str = Field(default='public')
     srid: int = Field(default=4326)
     spire_token:str = Field(default='')
     data_folder:str=Field(default=str(Path(__file__).parent.parent.parent.joinpath('./data')))

backend/bloom/infra/database/sql_model.py

@@ -22,6 +22,7 @@
 
 class Vessel(Base):
     __tablename__ = "dim_vessel"
+    __table_args__ = {'schema': settings.postgres_schema}
     id = Column("id", Integer, primary_key=True)
     mmsi = Column("mmsi", Integer)
     ship_name = Column("ship_name", String, nullable=False)
@@ -47,6 +48,7 @@ class Vessel(Base):
 
 class Alert(Base):
     __tablename__ = "alert"
+    __table_args__ = {'schema': settings.postgres_schema}
     id = Column("id", Integer, primary_key=True, index=True)
     timestamp = Column("timestamp", DateTime)
     mpa_id = Column("mpa_id", Integer)
@@ -55,6 +57,7 @@ class Alert(Base):
 
 class Port(Base):
     __tablename__ = "dim_port"
+    __table_args__ = {'schema': settings.postgres_schema}
     id = Column("id", Integer, primary_key=True, index=True)
     name = Column("name", String, nullable=False)
     locode = Column("locode", String, nullable=False)
@@ -71,6 +74,7 @@ class Port(Base):
 
 class SpireAisData(Base):
     __tablename__ = "spire_ais_data"
+    __table_args__ = {'schema': settings.postgres_schema}
 
     id = Column("id", Integer, primary_key=True)
     spire_update_statement = Column("spire_update_statement", DateTime(timezone=True))
@@ -108,6 +112,7 @@ class SpireAisData(Base):
 
 class Zone(Base):
     __tablename__ = "dim_zone"
+    __table_args__ = {'schema': settings.postgres_schema}
     id = Column("id", Integer, primary_key=True)
     category = Column("category", String, nullable=False)
     sub_category = Column("sub_category", String)
@@ -121,6 +126,7 @@ class Zone(Base):
 
 class WhiteZone(Base):
     __tablename__ = "dim_white_zone"
+    __table_args__ = {'schema': settings.postgres_schema}
     id = Column("id", Integer, primary_key=True)
     geometry = Column("geometry", Geometry(geometry_type="GEOMETRY", srid=settings.srid))
     created_at = Column("created_at", DateTime(timezone=True), server_default=func.now())
@@ -129,6 +135,7 @@ class WhiteZone(Base):
 
 class VesselPosition(Base):
     __tablename__ = "vessel_positions"
+    __table_args__ = {'schema': settings.postgres_schema}
     id = Column("id", Integer, primary_key=True)
     timestamp = Column("timestamp", DateTime(timezone=True), nullable=False)
     accuracy = Column("accuracy", String)
@@ -148,6 +155,7 @@ class VesselPosition(Base):
 
 class VesselData(Base):
     __tablename__ = "vessel_data"
+    __table_args__ = {'schema': settings.postgres_schema}
     id = Column("id", Integer, primary_key=True)
     timestamp = Column("timestamp", DateTime(timezone=True), nullable=False)
     ais_class = Column("ais_class", String)
@@ -166,6 +174,7 @@ class VesselData(Base):
 
 class VesselVoyage(Base):
     __tablename__ = "vessel_voyage"
+    __table_args__ = {'schema': settings.postgres_schema}
     id = Column("id", Integer, primary_key=True)
     timestamp = Column("timestamp", DateTime(timezone=True), nullable=False)
     destination = Column("destination", String)
@@ -177,6 +186,7 @@ class VesselVoyage(Base):
 
 class Excursion(Base):
     __tablename__ = "fct_excursion"
+    __table_args__ = {'schema': settings.postgres_schema}
     id = Column("id", Integer, primary_key=True)
     vessel_id = Column("vessel_id", Integer, ForeignKey("dim_vessel.id"), nullable=False)
     departure_port_id = Column("departure_port_id", Integer, ForeignKey("dim_port.id"))
@@ -201,6 +211,7 @@ class Excursion(Base):
 
 class Segment(Base):
     __tablename__ = "fct_segment"
+    __table_args__ = {'schema': settings.postgres_schema}
     id = Column("id", Integer, primary_key=True)
     excursion_id = Column("excursion_id", Integer, ForeignKey("fct_excursion.id"), nullable=False)
     timestamp_start = Column("timestamp_start", DateTime(timezone=True))
@@ -226,6 +237,7 @@ class Segment(Base):
 
 class TaskExecution(Base):
     __tablename__ = "tasks_executions"
+    __table_args__ = {'schema': settings.postgres_schema}
     id = Column("id", Integer, Identity(), primary_key=True)
     task_name = Column("task_name", String)
     point_in_time = Column("point_in_time", DateTime(timezone=True))
@@ -241,6 +253,7 @@ class RelSegmentZone(Base):
     __tablename__ = "rel_segment_zone"
     __table_args__ = (
         PrimaryKeyConstraint('segment_id', 'zone_id'),
+        {'schema': settings.postgres_schema}
     )
     segment_id = Column("segment_id", Integer, ForeignKey("fct_segment.id"), nullable=False)
     zone_id = Column("zone_id", Integer, ForeignKey("dim_zone.id"), nullable=False)
@@ -260,6 +273,7 @@ class RelSegmentZone(Base):
 
 class MetricsVesselInActivity(Base):
     __table__ = vessel_in_activity_request
+    __table_args__ = {'schema': settings.postgres_schema}
     #vessel_id: Mapped[Optional[int]]
     #total_time_at_sea: Mapped[Optional[timedelta]]
 

backend/bloom/infra/repositories/repository_port.py

@@ -13,10 +13,12 @@
 from sqlalchemy.orm import Session
 from bloom.domain.excursion import Excursion
 
+from bloom.infra.repository import GenericRepository, GenericSqlRepository
+from abc import ABC, abstractmethod
 
-class PortRepository:
-    def __init__(self, session_factory: Callable) -> None:
-        self.session_factory = session_factory
+class PortRepositoryBase(GenericRepository[Port], ABC):
+    def get_empty_geometry_buffer_ports(self, session: Session) -> list[Port]:
+        raise NotImplementedError()
 
     def get_port_by_id(self, session: Session, port_id: int) -> Union[Port, None]:
         entity = session.get(sql_model.Port, port_id)
@@ -37,67 +39,51 @@ def get_empty_geometry_buffer_ports(self, session: Session) -> list[Port]:
         if not q:
             return []
         return [PortRepository.map_to_domain(entity) for entity in q]
-
-    def get_ports_updated_created_after(self, session: Session, created_updated_after: datetime) -> list[Port]:
-        stmt = select(sql_model.Port).where(or_(sql_model.Port.created_at >= created_updated_after,
-                                                sql_model.Port.updated_at >= created_updated_after))
-        q = session.execute(stmt).scalars()
-        if not q:
-            return []
-        return [PortRepository.map_to_domain(entity) for entity in q]
-
-    def update_geometry_buffer(self, session: Session, port_id: int, buffer: Polygon) -> None:
-        session.execute(update(sql_model.Port), [{"id": port_id, "geometry_buffer": from_shape(buffer)}])
-
-    def batch_update_geometry_buffer(self, session: Session, id_buffers: list[dict[str, Any]]) -> None:
-        items = [{"id": item["id"], "geometry_buffer": from_shape(item["geometry_buffer"])} for item in id_buffers]
-        session.execute(update(sql_model.Port), items)
-
-    def create_port(self, session: Session, port: Port) -> Port:
-        orm_port = PortRepository.map_to_sql(port)
-        session.add(orm_port)
-        return PortRepository.map_to_domain(orm_port)
-
-    def batch_create_port(self, session: Session, ports: list[Port]) -> list[Port]:
-        orm_list = [PortRepository.map_to_sql(port) for port in ports]
-        session.add_all(orm_list)
-        return [PortRepository.map_to_domain(orm) for orm in orm_list]
-
-    def find_port_by_position_in_port_buffer(self, session: Session, position: Point) -> Union[Port, None]:
-        stmt = select(sql_model.Port).where(
-            func.ST_contains(sql_model.Port.geometry_buffer, from_shape(position, srid=settings.srid)) == True)
-        port = session.execute(stmt).scalar()
-        if not port:
-            return None
-        else:
-            return PortRepository.map_to_domain(port)
-
-    def find_port_by_distance(self,
-                              session: Session,
-                              longitude: float,
-                              latitude: float,
-                              threshold_distance_to_port: float) -> Union[Port, None]:
-        position = Point(longitude, latitude)
-        stmt = select(sql_model.Port).where(
-            and_(
-                func.ST_within(from_shape(position, srid=settings.srid),
-                               sql_model.Port.geometry_buffer) == True,
-                func.ST_distance(from_shape(position, srid=settings.srid),
-                                 sql_model.Port.geometry_point) < threshold_distance_to_port
-            )
-        ).order_by(asc(func.ST_distance(from_shape(position, srid=settings.srid),
-                                        sql_model.Port.geometry_point)))
-        result = session.execute(stmt).scalars()
-        return [PortRepository.map_to_domain(e) for e in result]
-
-    def get_closest_port_in_range(self, session: Session, longitude: float, latitude: float, range: float) -> Union[
-        tuple[int, float], None]:
-        res = session.execute(text("""SELECT id,ST_Distance(ST_POINT(:longitude,:latitude, 4326)::geography, geometry_point::geography) 
-                                FROM dim_port WHERE ST_Within(ST_POINT(:longitude,:latitude, 4326),geometry_buffer) = true
-                                AND ST_Distance(ST_POINT(:longitude,:latitude, 4326)::geography, geometry_point::geography) < :range
-                                ORDER by ST_Distance(ST_POINT(:longitude,:latitude, 4326)::geography, geometry_point::geography) ASC LIMIT 1"""),
-                              {"longitude": longitude, "latitude": latitude, "range": range}).first()
-        return res
+    pass
+
+# class PortRepository:
+#     def __init__(self, session_factory: Callable) -> None:
+#         self.session_factory = session_factory
+
+#     def get_port_by_id(self, session: Session, port_id: int) -> Union[Port, None]:
+#         entity = session.get(sql_model.Port, port_id)
+#         if entity is not None:
+#             return PortRepository.map_to_domain(entity)
+#         else:
+#             return None
+
+#     def get_all_ports(self, session: Session) -> List[Port]:
+#         q = session.query(sql_model.Port)
+#         if not q:
+#             return []
+#         return [PortRepository.map_to_domain(entity) for entity in q]
+
+#     def get_empty_geometry_buffer_ports(self, session: Session) -> list[Port]:
+#         stmt = select(sql_model.Port).where(sql_model.Port.geometry_buffer.is_(None))
+#         q = session.execute(stmt).scalars()
+#         if not q:
+#             return []
+#         return [PortRepository.map_to_domain(entity) for entity in q]
+
+#     def get_ports_updated_created_after(self, session: Session, created_updated_after: datetime) -> list[Port]:
+#         stmt = select(sql_model.Port).where(or_(sql_model.Port.created_at >= created_updated_after,
+#                                                 sql_model.Port.updated_at >= created_updated_after))
+#         q = session.execute(stmt).scalars()
+#         if not q:
+#             return []
+#         return [PortRepository.map_to_domain(entity) for entity in q]
+
+#     def update_geometry_buffer(self, session: Session, port_id: int, buffer: Polygon) -> None:
+#         session.execute(update(sql_model.Port), [{"id": port_id, "geometry_buffer": from_shape(buffer)}])
+
+#     def batch_update_geometry_buffer(self, session: Session, id_buffers: list[dict[str, Any]]) -> None:
+#         items = [{"id": item["id"], "geometry_buffer": from_shape(item["geometry_buffer"])} for item in id_buffers]
+#         session.execute(update(sql_model.Port), items)
+
+#     def create_port(self, session: Session, port: Port) -> Port:
+#         orm_port = PortRepository.map_to_sql(port)
+#         session.add(orm_port)
+#         return PortRepository.map_to_domain(orm_port)
 
 
     def update_port_has_excursion(self, session : Session, port_id: int ):

backend/bloom/infra/repository.py

@@ -57,7 +57,7 @@ def _construct_get_stmt(self, id: int) -> ScalarSelect:
 
     def get_by_id(self, id: int) -> Optional[SCHEMA]:
         stmt = self._construct_get_stmt(id)
-        return self._session.execute(stmt).scalar_one_or_none()
+        return self.map_to_domain(self._session.execute(stmt).scalar_one_or_none())
 
     def _construct_list_stmt(self, **filters) -> ScalarSelect:
         stmt = select(self._model_cls)
@@ -75,7 +75,7 @@ def _construct_list_stmt(self, **filters) -> ScalarSelect:
 
     def list(self, **filters) -> List[SCHEMA]:
         stmt = self._construct_list_stmt(**filters)
-        return self._session.execute(stmt).scalars()
+        return [ self.map_to_domain(item) for item in self._session.execute(stmt).scalars()]
 
     def add(self, record: SCHEMA) -> SCHEMA:
         self._session.add(record)

backend/bloom/routers/v1/ports.py

@@ -35,7 +35,7 @@ async def get_port(port_id:int,
                         key: str = Depends(X_API_KEY_HEADER)):
     check_apikey(key)
     use_cases = UseCases()
-    port_repository = use_cases.port_repository()
     db = use_cases.db()
     with db.session() as session:
-        return port_repository.get_port_by_id(session,port_id)
+        port_repository = use_cases.port_repository(session)
+        return port_repository.get_by_id(port_id)

backend/bloom/routers/v1/vessels.py

@@ -12,7 +12,6 @@
 from fastapi.encoders import jsonable_encoder
 router = APIRouter()
 
-
 @router.get("/vessels/trackedCount")
 async def list_vessel_tracked(request: Request, # used by @cache
                        key: str = Depends(X_API_KEY_HEADER)):

backend/bloom/services/geo.py

@@ -29,10 +29,10 @@ def find_positions_in_port_buffer(vessel_positions: List[tuple]) -> List[tuple]:
 
     # Get all ports from DataBase
     use_cases = UseCases()
-    port_repository = use_cases.port_repository()
     db = use_cases.db()
     with db.session() as session:
-        ports = port_repository.get_all_ports(session)
+        port_repository = use_cases.port_repository(session)
+        ports = port_repository.list()
 
     df_ports = pd.DataFrame(
         [[p.id, p.name, p.geometry_buffer] for p in ports],

backend/pyproject.toml

@@ -46,6 +46,8 @@ dependencies = [
     "fastapi[standard]>=0.115.0,<1.0.0",
     "uvicorn~=0.32",
     "redis~=5.0",
+    "pytest>=8.3.3",
+    "pytest-env>=1.1.5",
 ]
 name = "bloom"
 version = "0.1.0"
@@ -157,3 +159,14 @@ target-version = "py310"
 [tool.ruff.mccabe]
 max-complexity = 10
 
+
+[tool.pytest.ini_options]
+env = [
+    "POSTGRES_DRIVER=sqlite",
+    "POSTGRES_USER=",
+    "POSTGRES_PASSWORD=",
+    "POSTGRES_HOSTNAME=",
+    "POSTGRES_PORT=",
+    "POSTGRES_DB=:memory:",
+]
+

docker/Dockerfile

@@ -23,9 +23,9 @@ COPY ./backend/ ${PROJECT_DIR}/backend
 COPY docker/rsyslog.conf /etc/rsyslog.conf
 
 # Install requirements package for python with poetry
-ARG POETRY_VERSION=1.8.2
-ENV POETRY_VERSION=${POETRY_VERSION}
-RUN pip install --upgrade pip && pip install --user "poetry==$POETRY_VERSION"
+#ARG POETRY_VERSION=1.8.2
+#ENV POETRY_VERSION=${POETRY_VERSION}
+#RUN pip install --upgrade pip && pip install --user "poetry==$POETRY_VERSION"
 ENV PATH="${PATH}:/root/.local/bin"
 COPY ./backend/pyproject.toml ./backend/alembic.ini ./backend/
 
@@ -37,7 +37,8 @@ ENV UV_PROJECT_ENVIRONMENT=${VIRTUAL_ENV}
 RUN \
     cd backend &&\
     uv venv ${VIRTUAL_ENV} &&\
-    echo ". ${VIRTUAL_ENV}/bin/activate" >> /root/.bashrc &&\
+    echo ". ${VIRTUAL_ENV}/bin/activate" >> ~/.bashrc &&\
+    . ${VIRTUAL_ENV}/bin/activate &&\
     uv sync
 
 # Launch cron services