Merge pull request #11 from OCHA-DAP/HAPI-132/food-security-schema

HAPI-132 food security

changelog.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.4.0]
+
+### Added
+
+- Food security-related tables and views
+
 ## [0.3.2]
 
 ### Changed

src/hapi_schema/db_food_security.py

@@ -0,0 +1,124 @@
+"""Population table and view."""
+from datetime import datetime
+
+from sqlalchemy import (
+    DateTime,
+    Float,
+    ForeignKey,
+    Integer,
+    Text,
+    select,
+    text,
+)
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from hapi_schema.db_admin1 import DBAdmin1
+from hapi_schema.db_admin2 import DBAdmin2
+from hapi_schema.db_dataset import DBDataset
+from hapi_schema.db_ipc_phase import DBIpcPhase
+from hapi_schema.db_ipc_type import DBIpcType
+from hapi_schema.db_location import DBLocation
+from hapi_schema.db_resource import DBResource
+from hapi_schema.utils.base import Base
+from hapi_schema.utils.view_params import ViewParams
+
+
+class DBFoodSecurity(Base):
+    __tablename__ = "food_security"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    resource_ref: Mapped[int] = mapped_column(
+        ForeignKey("resource.id", onupdate="CASCADE", ondelete="CASCADE"),
+        nullable=False,
+    )
+    admin2_ref: Mapped[int] = mapped_column(
+        ForeignKey("admin2.id", onupdate="CASCADE"), nullable=False
+    )
+    ipc_phase_code: Mapped[str] = mapped_column(
+        ForeignKey("ipc_phase.code", onupdate="CASCADE"), nullable=False
+    )
+    ipc_type_code: Mapped[str] = mapped_column(
+        ForeignKey("ipc_type.code", onupdate="CASCADE"), nullable=False
+    )
+    population_in_phase: Mapped[int] = mapped_column(Integer, nullable=False)
+    population_fraction_in_phase: Mapped[float] = mapped_column(
+        Float, nullable=False
+    )
+    reference_period_start: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False
+    )
+    reference_period_end: Mapped[datetime] = mapped_column(
+        DateTime, nullable=True, server_default=text("NULL")
+    )
+    source_data: Mapped[str] = mapped_column(Text, nullable=True)
+
+    resource = relationship("DBResource")
+    admin2 = relationship("DBAdmin2")
+    ipc_phase = relationship("DBIpcPhase")
+    ipc_type = relationship("DBIpcType")
+
+
+view_params_food_security = ViewParams(
+    name="food_security_view",
+    metadata=Base.metadata,
+    selectable=select(
+        *DBFoodSecurity.__table__.columns,
+        DBDataset.hdx_id.label("dataset_hdx_id"),
+        DBDataset.hdx_stub.label("dataset_hdx_stub"),
+        DBDataset.title.label("dataset_title"),
+        DBDataset.hdx_provider_stub.label("dataset_hdx_provider_stub"),
+        DBDataset.hdx_provider_name.label("dataset_hdx_provider_name"),
+        DBIpcPhase.name.label("ipc_phase_name"),
+        DBResource.hdx_id.label("resource_hdx_id"),
+        DBResource.name.label("resource_name"),
+        DBResource.update_date.label("resource_update_date"),
+        DBLocation.code.label("location_code"),
+        DBLocation.name.label("location_name"),
+        DBAdmin1.code.label("admin1_code"),
+        DBAdmin1.name.label("admin1_name"),
+        DBAdmin1.is_unspecified.label("admin1_is_unspecified"),
+        DBAdmin2.code.label("admin2_code"),
+        DBAdmin2.name.label("admin2_name"),
+        DBAdmin2.is_unspecified.label("admin2_is_unspecified")
+    ).select_from(
+        # Join pop to admin2 to admin1 to loc
+        DBFoodSecurity.__table__.join(
+            DBAdmin2.__table__,
+            DBFoodSecurity.admin2_ref == DBAdmin2.id,
+            isouter=True,
+        )
+        .join(
+            DBAdmin1.__table__,
+            DBAdmin2.admin1_ref == DBAdmin1.id,
+            isouter=True,
+        )
+        .join(
+            DBLocation.__table__,
+            DBAdmin1.location_ref == DBLocation.id,
+            isouter=True,
+        )
+        # Join pop to resource to dataset
+        .join(
+            DBResource.__table__,
+            DBFoodSecurity.resource_ref == DBResource.id,
+            isouter=True,
+        )
+        .join(
+            DBDataset.__table__,
+            DBResource.dataset_ref == DBDataset.id,
+            isouter=True,
+        )
+        # Join to ipc phase
+        .join(
+            DBIpcPhase.__table__,
+            DBFoodSecurity.ipc_phase_code == DBIpcPhase.code,
+            isouter=True,
+        )
+        # Join to ipc type
+        .join(
+            DBIpcType.__table__,
+            DBFoodSecurity.ipc_type_code == DBIpcType.code,
+            isouter=True,
+        )
+    ),
+)

src/hapi_schema/db_ipc_phase.py

@@ -0,0 +1,22 @@
+"""IPC phase table and view."""
+
+from sqlalchemy import String, select
+from sqlalchemy.orm import Mapped, mapped_column
+
+from hapi_schema.utils.base import Base
+from hapi_schema.utils.view_params import ViewParams
+
+
+class DBIpcPhase(Base):
+    __tablename__ = "ipc_phase"
+
+    code: Mapped[int] = mapped_column(String(32), primary_key=True)
+    name: Mapped[str] = mapped_column(String(32), nullable=False)
+    description: Mapped[str] = mapped_column(String(512), nullable=False)
+
+
+view_params_ipc_phase = ViewParams(
+    name="ipc_phase_view",
+    metadata=Base.metadata,
+    selectable=select(*DBIpcPhase.__table__.columns),
+)

src/hapi_schema/db_ipc_type.py

@@ -0,0 +1,21 @@
+"""IPC type table and view."""
+
+from sqlalchemy import String, select
+from sqlalchemy.orm import Mapped, mapped_column
+
+from hapi_schema.utils.base import Base
+from hapi_schema.utils.view_params import ViewParams
+
+
+class DBIpcType(Base):
+    __tablename__ = "ipc_type"
+
+    code: Mapped[str] = mapped_column(String(32), primary_key=True)
+    description: Mapped[str] = mapped_column(String(512), nullable=False)
+
+
+view_params_ipc_type = ViewParams(
+    name="ipca_type_vew",
+    metadata=Base.metadata,
+    selectable=select(*DBIpcType.__table__.columns),
+)

src/hapi_schema/utils/view_params.py

@@ -1,13 +1,13 @@
 from dataclasses import dataclass
 
-from sqlalchemy.orm import DeclarativeMeta
 from sqlalchemy.sql.expression import Selectable
+from sqlalchemy.sql.schema import MetaData
 
 
 @dataclass
 class ViewParams:
     """Class for keeping view constructor parameters."""
 
     name: str
-    metadata: DeclarativeMeta
+    metadata: MetaData
     selectable: Selectable

tests/conftest.py

@@ -6,7 +6,10 @@
 from hapi_schema.db_admin2 import DBAdmin2
 from hapi_schema.db_age_range import DBAgeRange
 from hapi_schema.db_dataset import DBDataset
+from hapi_schema.db_food_security import DBFoodSecurity
 from hapi_schema.db_gender import DBGender
+from hapi_schema.db_ipc_phase import DBIpcPhase
+from hapi_schema.db_ipc_type import DBIpcType
 from hapi_schema.db_location import DBLocation
 from hapi_schema.db_operational_presence import (
     DBOperationalPresence,
@@ -21,7 +24,10 @@
 from sample_data.data_admin2 import data_admin2
 from sample_data.data_age_range import data_age_range
 from sample_data.data_dataset import data_dataset
+from sample_data.data_food_security import data_food_security
 from sample_data.data_gender import data_gender
+from sample_data.data_ipc_phase import data_ipc_phase
+from sample_data.data_ipc_type import data_ipc_type
 from sample_data.data_location import data_location
 from sample_data.data_operational_presence import data_operational_presence
 from sample_data.data_org import data_org
@@ -44,7 +50,10 @@ def engine():
     session.execute(insert(DBAdmin2), data_admin2)
     session.execute(insert(DBAgeRange), data_age_range)
     session.execute(insert(DBDataset), data_dataset)
+    session.execute(insert(DBFoodSecurity), data_food_security)
     session.execute(insert(DBGender), data_gender)
+    session.execute(insert(DBIpcPhase), data_ipc_phase)
+    session.execute(insert(DBIpcType), data_ipc_type)
     session.execute(insert(DBLocation), data_location)
     session.execute(insert(DBOperationalPresence), data_operational_presence)
     session.execute(insert(DBOrg), data_org)

tests/sample_data/data_food_security.py

@@ -0,0 +1,46 @@
+from datetime import datetime
+
+data_food_security = [
+    # Phase 1 current national
+    dict(
+        id=1,
+        resource_ref=3,
+        admin2_ref=1,
+        ipc_phase_code=1,
+        ipc_type_code=1,
+        population_total=1_000_000,
+        population_in_phase=500_000,
+        population_fraction_in_phase=0.5,
+        reference_period_start=datetime(2023, 1, 1),
+        reference_period_end=datetime(2023, 3, 30),
+        source_data="DATA,DATA,DATA",
+    ),
+    # Phase 2 first projection admin1
+    dict(
+        id=2,
+        resource_ref=3,
+        admin2_ref=2,
+        ipc_phase_code=2,
+        ipc_type_code=2,
+        population_total=100_000,
+        population_in_phase=40_000,
+        population_fraction_in_phase=0.4,
+        reference_period_start=datetime(2023, 4, 1),
+        reference_period_end=datetime(2023, 6, 30),
+        source_data="DATA,DATA,DATA",
+    ),
+    # Phase 3 second projection admin2
+    dict(
+        id=3,
+        resource_ref=3,
+        admin2_ref=4,
+        ipc_phase_code=3,
+        ipc_type_code=3,
+        population_total=10_000,
+        population_in_phase=3_000,
+        population_fraction_in_phase=0.3,
+        reference_period_start=datetime(2023, 7, 1),
+        reference_period_end=datetime(2023, 10, 31),
+        source_data="DATA,DATA,DATA",
+    ),
+]

tests/sample_data/data_ipc_phase.py

@@ -0,0 +1,35 @@
+# Taken from page 53 of
+# https://www.ipcinfo.org/fileadmin/user_upload/ipcinfo/manual/IPC_Technical_Manual_3_Final.pdf
+data_ipc_phase = [
+    dict(
+        code="1",
+        name="Phase 1: None/Minimal",
+        description="Households are able to meet essential food and non-food "
+        "needs without engaging in atypical and unsustainable strategies to "
+        "access food and income.",
+    ),
+    dict(
+        code="2",
+        name="Phase 2: Stressed",
+        description="Households have minimally adequate food consumption but "
+        "are unable to afford some essential non-food expenditures without "
+        "engaging in stress-coping strategies.",
+    ),
+    dict(
+        code="3",
+        name="Phase 3: Crisis",
+        description="Households either have food consumption gaps that are "
+        "reflected by high or above-usual acute malnutrition, or are "
+        "marginally able to meet minimum food needs but only by depleting "
+        "essential livelihood assets or through crisis-coping strategies.",
+    ),
+    dict(
+        code="3+",
+        name="Phase 3+: In Need of Action",
+        description="Sum of population in phases 3, 4, and 5. The population "
+        "in Phase 3+ does not necessarily reflect the full population in need "
+        "of urgent action. This is because some households may be in Phase 2 "
+        "or even 1 but only because of receipt of assistance, and thus, they "
+        "may be in need of continued action.",
+    ),
+]

tests/sample_data/data_ipc_type.py

@@ -0,0 +1,14 @@
+data_ipc_type = [
+    dict(
+        code="current",
+        description="Food insecurity that is occurring in the current analysis period.",
+    ),
+    dict(
+        code="first projection",
+        description="Projected food insecurity occurring in the period immediately following the current analysis period.",
+    ),
+    dict(
+        code="second projection",
+        description="Projected food insecurity occurring in the period immediately following the first projection period.",
+    ),
+]