feat: add create_gtfs_feed endpoint to operations API (#1427)

Author: davidgamez

.gitignore

@@ -83,4 +83,7 @@ functions-python/**/*.csv
 .cloudstorage
 
 # Project files
-*.code-workspace
+*.code-workspace
+
+# Ignore OpenApi local backup files
+*.yaml.bak

api/src/shared/common/locations_utils.py

@@ -0,0 +1,110 @@
+import logging
+from typing import Optional
+
+from sqlalchemy.orm import Session
+
+from shared.database_gen.sqlacodegen_models import Location
+
+
+def get_country_code(country_name: str) -> Optional[str]:
+    """
+    Get ISO 3166 country code from country name
+
+    Args:
+        country_name (str): Full country name
+
+    Returns:
+        Optional[str]: Two-letter ISO country code or None if not found
+    """
+    import pycountry
+
+    # Return None for empty or whitespace-only strings
+    if not country_name or not country_name.strip():
+        logging.error("Could not find country code for: empty string")
+        return None
+
+    try:
+        # Try exact match first
+        country = pycountry.countries.get(name=country_name)
+        if country:
+            return country.alpha_2
+
+        # Try searching with fuzzy matching
+        countries = pycountry.countries.search_fuzzy(country_name)
+        if countries:
+            return countries[0].alpha_2
+
+    except LookupError:
+        logging.error(f"Could not find country code for: {country_name}")
+    return None
+
+
+def create_or_get_location(
+    session: Session,
+    country: Optional[str],
+    state_province: Optional[str],
+    city_name: Optional[str],
+    country_code: Optional[str] = None,
+) -> Optional[Location]:
+    """
+    Create a new location or get existing one
+
+    Args:
+        session: Database session
+        country: Country name
+        state_province: State/province name
+        city_name: City name
+        country_code: ISO 3166-1 alpha-2 country code
+
+    Returns:
+        Optional[Location]: Location object or None if creation failed
+    """
+    import pycountry
+
+    if not any([country, state_province, city_name]):
+        return None
+
+    # Generate location_id using the specified pattern
+    location_components = []
+    if country_code:
+        location_components.append(country_code)
+        try:
+            py_country = pycountry.countries.get(alpha_2=country_code.strip())
+            country = py_country.name if py_country else country
+        except LookupError:
+            logging.warning("Could not find country code for: %s", country_code)
+    elif country:
+        country_code = get_country_code(country)
+        if country_code:
+            location_components.append(country_code)
+        else:
+            logging.error(f"Could not determine country code for {country}")
+            return None
+
+    if state_province:
+        location_components.append(state_province)
+    if city_name:
+        location_components.append(city_name)
+
+    location_id = "-".join(location_components)
+
+    # First check if location already exists
+    existing_location = session.query(Location).filter(Location.id == location_id).first()
+
+    if existing_location:
+        logging.debug(f"Using existing location: {location_id}")
+        return existing_location
+
+    # Create new location
+    location = Location(
+        id=location_id,
+        country_code=country_code,
+        country=country,
+        subdivision_name=state_province,
+        municipality=city_name,
+    )
+    session.add(location)
+    session.flush()
+    logging.debug(f"Created new location: {location_id}")
+
+    return location

api/src/shared/db_models/external_id_impl.py

@@ -21,3 +21,12 @@ def from_orm(cls, external_id: Externalid | None) -> ExternalId | None:
             external_id=external_id.associated_id,
             source=external_id.source,
         )
+
+    @classmethod
+    def to_orm_from_dict(cls, external_id_dict: dict | None) -> Externalid | None:
+        if not external_id_dict:
+            return None
+        return Externalid(
+            associated_id=external_id_dict.get("external_id"),
+            source=external_id_dict.get("source"),
+        )

api/src/shared/db_models/feed_impl.py

@@ -1,8 +1,12 @@
 from shared.db_models.basic_feed_impl import BaseFeedImpl
 from feeds_gen.models.feed import Feed
 from shared.database_gen.sqlacodegen_models import Feed as FeedOrm
+from shared.db_models.external_id_impl import ExternalIdImpl
 from shared.db_models.feed_related_link_impl import FeedRelatedLinkImpl
 
+from shared.db_models.location_impl import LocationImpl
+from shared.db_models.redirect_impl import RedirectImpl
+
 
 class FeedImpl(BaseFeedImpl, Feed):
     """Base implementation of the feeds models.
@@ -27,3 +31,48 @@ def from_orm(cls, feed_orm: FeedOrm | None) -> Feed | None:
         feed.related_links = [FeedRelatedLinkImpl.from_orm(related_link) for related_link in feed_orm.feedrelatedlinks]
         feed.note = feed_orm.note
         return feed
+
+    @classmethod
+    def to_orm_from_dict(cls, feed_dict: dict | None) -> FeedOrm | None:
+        """Convert a dictionary representation of a feed to a SQLAlchemy Feed ORM object."""
+        if not feed_dict:
+            return None
+        result: Feed = FeedOrm(
+            id=feed_dict.get("id"),
+            stable_id=feed_dict.get("stable_id"),
+            data_type=feed_dict.get("data_type"),
+            created_at=feed_dict.get("created_at"),
+            provider=feed_dict.get("provider"),
+            feed_contact_email=feed_dict.get("feed_contact_email"),
+            producer_url=feed_dict.get("producer_url"),
+            authentication_type=None
+            if feed_dict.get("authentication_type") is None
+            else str(feed_dict.get("authentication_type")),
+            authentication_info_url=feed_dict.get("authentication_info_url"),
+            api_key_parameter_name=feed_dict.get("api_key_parameter_name"),
+            license_url=feed_dict.get("license_url"),
+            status=feed_dict.get("status"),
+            official=feed_dict.get("official"),
+            official_updated_at=feed_dict.get("official_updated_at"),
+            feed_name=feed_dict.get("feed_name"),
+            note=feed_dict.get("note"),
+            externalids=sorted(
+                [ExternalIdImpl.to_orm_from_dict(item) for item in feed_dict.get("externalids")],
+                key=lambda x: x.associated_id,
+            )
+            if feed_dict.get("externalids")
+            else [],
+            redirectingids=sorted(
+                [RedirectImpl.to_orm_from_dict(item) for item in feed_dict.get("redirectingids")],
+                key=lambda x: x.target_id,
+            )
+            if feed_dict.get("redirectingids")
+            else [],
+            feedrelatedlinks=[FeedRelatedLinkImpl.to_orm_from_dict(item) for item in feed_dict.get("feedrelatedlinks")]
+            if feed_dict.get("feedrelatedlinks")
+            else [],
+            locations=[LocationImpl.to_orm_from_dict(item) for item in feed_dict.get("locations")]
+            if feed_dict.get("locations")
+            else [],
+        )
+        return result

api/src/shared/db_models/feed_related_link_impl.py

@@ -21,3 +21,15 @@ def from_orm(cls, feed_related_link_orm: Feedrelatedlink) -> FeedRelatedLink | N
             description=feed_related_link_orm.description,
             created_at=feed_related_link_orm.created_at,
         )
+
+    @classmethod
+    def to_orm_from_dict(cls, feedrelaticlink_dict: dict) -> Feedrelatedlink | None:
+        """Convert a dict to a SQLAlchemy row object."""
+        if not feedrelaticlink_dict:
+            return None
+        result = Feedrelatedlink(
+            code=feedrelaticlink_dict.get("code"),
+            url=feedrelaticlink_dict.get("url"),
+            description=feedrelaticlink_dict.get("description"),
+        )
+        return result

api/src/shared/db_models/location_impl.py

@@ -1,5 +1,10 @@
+from sqlalchemy.orm import Session
+
 from feeds_gen.models.location import Location
 import pycountry
+
+from shared.common.locations_utils import create_or_get_location
+from shared.database.database import with_db_session
 from shared.database_gen.sqlacodegen_models import Location as LocationOrm
 
 
@@ -27,3 +32,15 @@ def from_orm(cls, location: LocationOrm | None) -> Location | None:
             subdivision_name=location.subdivision_name,
             municipality=location.municipality,
         )
+
+    @classmethod
+    @with_db_session
+    def to_orm_from_dict(self, location_dict: dict, db_session: Session) -> LocationOrm | None:
+        """Convert the Pydantic model instance to a SQLAlchemy Location row object."""
+        return create_or_get_location(
+            session=db_session,
+            country=location_dict["country"],
+            state_province=location_dict["subdivision_name"],
+            city_name=location_dict["municipality"],
+            country_code=location_dict["country_code"],
+        )

api/src/shared/db_models/redirect_impl.py

@@ -1,4 +1,7 @@
-from shared.database_gen.sqlacodegen_models import Redirectingid
+from sqlalchemy.orm import Session
+
+from shared.database.database import with_db_session
+from shared.database_gen.sqlacodegen_models import Redirectingid, Feed
 from feeds_gen.models.redirect import Redirect
 
 
@@ -21,3 +24,18 @@ def from_orm(cls, redirect: Redirectingid | None) -> Redirect | None:
             target_id=redirect.target.stable_id,
             comment=redirect.redirect_comment,
         )
+
+    @classmethod
+    @with_db_session
+    def to_orm_from_dict(cls, redirect_dict: dict | None, db_session: Session = None) -> Redirectingid | None:
+        # Return None if no payload or missing target_id
+        if not redirect_dict or redirect_dict.get("target_id") is None:
+            return None
+        target = db_session.query(Feed).filter_by(stable_id=redirect_dict.get("target_id")).first()
+        if not target:
+            return None
+        return Redirectingid(
+            target_id=target.id,
+            target=target,
+            redirect_comment=redirect_dict.get("comment"),
+        )

api/tests/unittest/models/test_basic_feed_impl.py

@@ -62,7 +62,7 @@
             feed_id="feed_id",
             hosted_url="hosted_url",
             note="note",
-            downloaded_at="downloaded_at",
+            downloaded_at=datetime(2023, 1, 1, 0, 0, 0, tzinfo=ZoneInfo("UTC")),
             hash="hash",
             bounding_box="bounding_box",
             service_date_range_start=datetime(2024, 1, 1, 0, 0, 0, tzinfo=ZoneInfo("Canada/Atlantic")),
@@ -160,3 +160,97 @@ def test_from_orm_empty_fields(self):
     def test_from_orm_none(self):
         """Test the `from_orm` method with None."""
         assert FeedImpl.from_orm(None) is None
+
+    def test_to_orm_from_dict_none(self):
+        """to_orm_from_dict returns None when input is None or empty dict."""
+        assert FeedImpl.to_orm_from_dict(None) is None
+        assert FeedImpl.to_orm_from_dict({}) is None
+
+    def test_to_orm_from_dict_full_payload(self):
+        """to_orm_from_dict maps primitives and nested collections; sorts externalids by associated_id."""
+        now = datetime(2024, 5, 1, 12, 30, 0)
+        updated = datetime(2024, 6, 1, 8, 0, 0)
+        payload = {
+            "id": "feed-123",
+            "stable_id": "stable-123",
+            "data_type": "gtfs",
+            "created_at": now,
+            "provider": "Provider A",
+            "feed_contact_email": "[email protected]",
+            "producer_url": "https://producer.example.com",
+            "authentication_type": 1,  # should be converted to string
+            "authentication_info_url": "https://auth.example.com",
+            "api_key_parameter_name": "api_key",
+            "license_url": "https://license.example.com",
+            "status": "active",
+            "official": True,
+            "official_updated_at": updated,
+            "feed_name": "Feed Name",
+            "note": "Some note",
+            # avoid DB-dependent fields: locations and redirectingids
+            "externalids": [
+                {"external_id": "b-id", "source": "src"},
+                {"external_id": "a-id", "source": "src"},
+            ],
+            "feedrelatedlinks": [
+                {"code": "docs", "url": "https://docs.example.com", "description": "Docs"},
+                {"code": "home", "url": "https://home.example.com", "description": "Home"},
+            ],
+        }
+
+        obj = FeedImpl.to_orm_from_dict(payload)
+
+        # Basic type
+        assert isinstance(obj, Feed)
+
+        # Primitives
+        assert obj.id == "feed-123"
+        assert obj.stable_id == "stable-123"
+        assert obj.data_type == "gtfs"
+        assert obj.created_at == now
+        assert obj.provider == "Provider A"
+        assert obj.feed_contact_email == "[email protected]"
+        assert obj.producer_url == "https://producer.example.com"
+        # authentication_type coerced to string per implementation
+        assert obj.authentication_type == "1"
+        assert obj.authentication_info_url == "https://auth.example.com"
+        assert obj.api_key_parameter_name == "api_key"
+        assert obj.license_url == "https://license.example.com"
+        assert obj.status == "active"
+        assert obj.official is True
+        assert obj.official_updated_at == updated
+        assert obj.feed_name == "Feed Name"
+        assert obj.note == "Some note"
+
+        # Nested: externalids should be sorted by associated_id
+        assert [type(e).__name__ for e in obj.externalids] == ["Externalid", "Externalid"]
+        got_ext = [(e.source, e.associated_id) for e in obj.externalids]
+        assert got_ext == [("src", "a-id"), ("src", "b-id")]
+
+        # Nested: feedrelatedlinks preserved order (no explicit sort in impl)
+        assert len(obj.feedrelatedlinks) == 2
+        codes = [feedrelatedlinks.code for feedrelatedlinks in obj.feedrelatedlinks]
+        urls = [feedrelatedlinks.url for feedrelatedlinks in obj.feedrelatedlinks]
+        assert codes == ["docs", "home"]
+        assert urls == ["https://docs.example.com", "https://home.example.com"]
+
+        # Relationships not provided should be empty lists
+        assert obj.redirectingids == []
+        assert obj.locations == []
+
+    def test_to_orm_from_dict_empty_collections(self):
+        """Explicit empty lists yield empty relationship collections in ORM object."""
+        payload = {
+            "stable_id": "s",
+            "data_type": "gtfs",
+            "externalids": [],
+            "feedrelatedlinks": [],
+            "redirectingids": [],
+            "locations": [],
+        }
+        obj = FeedImpl.to_orm_from_dict(payload)
+        assert isinstance(obj, Feed)
+        assert obj.externalids == []
+        assert obj.feedrelatedlinks == []
+        assert obj.redirectingids == []
+        assert obj.locations == []