Refactor to general city class (#342)

app/cities/__init__.py

@@ -1 +1,22 @@
 """General class for cities."""
+
+
+class City:
+    """General class for cities."""
+
+    def __init__(  # noqa: PLR0913
+        self,
+        name: str,
+        country: str,
+        country_id: int,
+        province_id: int,
+        geo_code: str,
+        phone_code: str,
+    ) -> None:
+        """Initialize the class."""
+        self.name = name
+        self.country = country
+        self.country_id = country_id
+        self.province_id = province_id
+        self.geo_code = geo_code
+        self.phone_code = phone_code

app/cities/germany/__init__.py

@@ -0,0 +1 @@
+"""Code for the cities in Germany."""

app/cities/germany/hamburg.py

@@ -0,0 +1,90 @@
+"""Python script for Park and Ride Hamburg data."""
+import datetime
+import json
+
+import pymysql
+import pytz
+from hamburg import ParkAndRide, UDPHamburg
+
+from app.cities import City
+from app.database import connection, cursor
+from app.helpers import get_unique_number
+
+
+class Municipality(City):
+    """Manage the location data of Hamburg."""
+
+    def __init__(self) -> None:
+        """Initialize the class."""
+        super().__init__(
+            name="Hamburg",
+            country="Germany",
+            country_id=83,
+            province_id=14,
+            geo_code="DE-HH",
+            phone_code="040",
+        )
+        self.bulk = "true"
+
+    async def async_get_locations(self) -> ParkAndRide:
+        """Get parking data from API.
+
+        Args:
+        ----
+            bulk (str): Get all data in one request.
+        """
+        async with UDPHamburg() as client:
+            parking: ParkAndRide = await client.park_and_rides(bulk=self.bulk)
+            print(f"{self.name} - data has been retrieved")
+            return parking
+
+    def upload_data(self, data_set: list, time: datetime) -> None:
+        """Update the database with new data.
+
+        Args:
+        ----
+            data_set (list): List of garages.
+            time (datetime): Current time.
+        """
+        # purge_database(municipality, time)  # noqa: ERA001
+        print(f"{time} - {self.name}: START updating database with new data")
+        try:
+            connection.ping(reconnect=True)
+            for item in data_set:
+                location_id = f"{self.geo_code}-{self.phone_code}-{get_unique_number(item.latitude, item.longitude)}"  # noqa: E501
+                sql = """INSERT INTO `parking_offstreet` (id, name, country_id, province_id, municipality, free_space_short, short_capacity, availability_pct, parking_type, prices, url, longitude, latitude, visibility, created_at, updated_at)
+                        VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY
+                        UPDATE id=values(id),
+                                name=values(name),
+                                state=values(state),
+                                free_space_short=values(free_space_short),
+                                short_capacity=values(short_capacity),
+                                availability_pct=values(availability_pct),
+                                prices=values(prices),
+                                longitude=values(longitude),
+                                latitude=values(latitude),
+                                updated_at=values(updated_at)"""  # noqa: E501
+                val = (
+                    location_id,
+                    str(item.name),
+                    int(self.country_id),
+                    int(self.province_id),
+                    str(self.name),
+                    item.free_space,
+                    item.capacity,
+                    item.availability_pct,
+                    "parkandride",
+                    json.dumps(item.tickets),
+                    item.url,
+                    float(item.longitude),
+                    float(item.latitude),
+                    bool(True),
+                    datetime.datetime.now(tz=pytz.timezone("Europe/Berlin")),
+                    item.updated_at,
+                )
+                cursor.execute(sql, val)
+            connection.commit()
+        except pymysql.Error as error:
+            print(f"MySQL error: {error}")
+        finally:
+            print(f"{time} - {self.name}: DONE with database update")

app/cities/netherlands/__init__.py

@@ -0,0 +1 @@
+"""Code for the cities in The Netherlands."""

app/cities/netherlands/amsterdam.py

@@ -0,0 +1,102 @@
+"""Python script for Garages Amsterdam data."""
+import datetime
+
+import pymysql
+import pytz
+from odp_amsterdam import Garage, ODPAmsterdam
+
+from app.cities import City
+from app.database import connection, cursor
+from app.helpers import get_unique_number
+
+
+class Municipality(City):
+    """Manage the location data of Amsterdam."""
+
+    def __init__(self) -> None:
+        """Initialize the class."""
+        super().__init__(
+            name="Amsterdam",
+            country="Netherlands",
+            country_id=157,
+            province_id=8,
+            geo_code="NL-NH",
+            phone_code="020",
+        )
+
+    async def async_get_locations(self) -> Garage:
+        """Get garage data from API.
+
+        Returns
+        -------
+            list: List of garages.
+        """
+        async with ODPAmsterdam() as client:
+            garages: Garage = await client.all_garages()
+            print(f"{self.name} - data has been retrieved")
+            return garages
+
+    def upload_data(self, data_set: list, time: datetime) -> None:
+        """Update the database with new data.
+
+        Args:
+        ----
+            data_set (list): List of garages.
+            time (datetime): Current time.
+        """
+        # purge_database(municipality, time)  # noqa: ERA001
+        print(f"{time} - {self.name}: START updating database with new data")
+        try:
+            connection.ping(reconnect=True)
+            for item in data_set:
+                location_id = f"{self.geo_code}-{self.phone_code}-{get_unique_number(item.latitude, item.longitude)}"  # noqa: E501
+                sql = """INSERT INTO `parking_offstreet` (id, name, country_id, province_id, municipality, state, free_space_short, free_space_long, short_capacity, long_capacity, availability_pct, parking_type, longitude, latitude, visibility, created_at, updated_at)
+                        VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY
+                        UPDATE id=values(id),
+                                name=values(name),
+                                state=values(state),
+                                free_space_short=values(free_space_short),
+                                free_space_long=values(free_space_long),
+                                short_capacity=values(short_capacity),
+                                long_capacity=values(long_capacity),
+                                availability_pct=values(availability_pct),
+                                longitude=values(longitude),
+                                latitude=values(latitude),
+                                updated_at=values(updated_at)"""  # noqa: E501
+                val = (
+                    location_id,
+                    str(item.garage_name),
+                    int(self.country_id),
+                    int(self.province_id),
+                    str(self.name),
+                    str(item.state),
+                    check_value(item.free_space_short),
+                    check_value(item.free_space_long),
+                    check_value(item.short_capacity),
+                    check_value(item.long_capacity),
+                    item.availability_pct,
+                    "garage",
+                    float(item.longitude),
+                    float(item.latitude),
+                    bool(True),
+                    datetime.datetime.now(tz=pytz.timezone("Europe/Amsterdam")),
+                    datetime.datetime.now(tz=pytz.timezone("Europe/Amsterdam")),
+                )
+                cursor.execute(sql, val)
+            connection.commit()
+        except pymysql.Error as error:
+            print(f"MySQL error: {error}")
+        finally:
+            print(f"{time} - {self.name}: DONE with database update")
+
+
+def check_value(value: int) -> int:
+    """Replace None values with 0.
+
+    Args:
+    ----
+        value (int): Value to check.
+    """
+    if value is None:
+        return 0
+    return value