Merge pull request #252 from GispoCoding/15-add-lisatiedon-lajit

15 add lisatiedon lajit

database/base.py

@@ -59,7 +59,11 @@ class VersionedBase(Base):
     __abstract__ = True
     __table_args__ = {"schema": "hame"}
 
-    id: Mapped[uuid_pk]
+    # Go figure. We have to *explicitly state* id is a mapped column, because id will
+    # have to be defined inside all the subclasses for relationship remote_side
+    # definition to work. So even if there is an id field in all the classes,
+    # self-relationships will later break if id is only defined by type annotation.
+    id: Mapped[uuid_pk] = mapped_column()
     created_at: Mapped[timestamp]
     # TODO: postgresql has no default onupdate. Must implement this with trigger.
     modified_at: Mapped[timestamp]
@@ -96,10 +100,12 @@ def parent_id(cls) -> Mapped[Optional[uuid.UUID]]:  # noqa
             ForeignKey(cls.id, name=f"{cls.__tablename__}_parent_id_fkey"), index=True
         )
 
-    @classmethod
+    # Oh great. Unlike SQLAlchemy documentation states, @classmethod decorator should
+    # absolutely *not* be used. Declared relationships are not correctly set if the
+    # decorator is present.
     @declared_attr
-    def parent(cls) -> Mapped[VersionedBase]:
-        return relationship(cls, backref="children")
+    def parent(cls) -> Mapped[Optional[VersionedBase]]:  # noqa
+        return relationship(cls, remote_side=[cls.id], backref="children")
 
     @property
     def uri(self):
@@ -124,9 +130,8 @@ class PlanBase(VersionedBase):
     )
 
     # class reference in abstract base class, with backreference to class name:
-    @classmethod
     @declared_attr
-    def lifecycle_status(cls) -> Mapped[VersionedBase]:
+    def lifecycle_status(cls) -> Mapped[VersionedBase]:  # noqa
         return relationship("LifeCycleStatus", backref=f"{cls.__tablename__}s")
 
 
@@ -156,19 +161,16 @@ class PlanObjectBase(PlanBase):
     )
 
     # class reference in abstract base class, with backreference to class name:
-    @classmethod
     @declared_attr
-    def type_of_underground(cls) -> Mapped[VersionedBase]:
+    def type_of_underground(cls) -> Mapped[VersionedBase]:  # noqa
         return relationship("TypeOfUnderground", backref=f"{cls.__tablename__}s")
 
     # class reference in abstract base class, with backreference to class name:
-    @classmethod
     @declared_attr
-    def plan(cls) -> Mapped[VersionedBase]:
+    def plan(cls) -> Mapped[VersionedBase]:  # noqa
         return relationship("Plan", backref=f"{cls.__tablename__}s")
 
     # class reference in abstract base class, with backreference to class name:
-    @classmethod
     @declared_attr
-    def plan_regulation_group(cls) -> Mapped[VersionedBase]:
+    def plan_regulation_group(cls) -> Mapped[VersionedBase]:  # noqa
         return relationship("PlanRegulationGroup", backref=f"{cls.__tablename__}s")

database/codes.py

@@ -28,7 +28,7 @@ class TypeOfPlanRegulation(CodeBase):
     code_list_uri = "http://uri.suomi.fi/codelist/rytj/RY_Kaavamaarayslaji"
 
 
-class TypeOfAdditionalInformationForPlanRegulation(CodeBase):
+class TypeOfAdditionalInformation(CodeBase):
     """
     Kaavamääräyksen lisätiedon laji
 
@@ -46,6 +46,38 @@ class TypeOfAdditionalInformationForPlanRegulation(CodeBase):
     code_list_uri = (
         "http://uri.suomi.fi/codelist/rytj/RY_Kaavamaarayksen_Lisatiedonlaji"
     )
+    local_codes = [
+        {
+            "value": "kayttotarkoitus",
+            "name": {"fin": "Käyttötarkoitus"},
+            "child_values": [
+                "paakayttotarkoitus",
+                "osaAlue",
+                "poisluettavaKayttotarkoitus",
+                "yhteystarve",
+            ],
+        },
+        {
+            "value": "olemassaolo",
+            "name": {"fin": "Olemassaolo"},
+            "child_values": [
+                "olemassaOleva",
+                "sailytettava",
+                "uusi",
+                "olennaisestiMuuttuva",
+            ],
+        },
+        {
+            "value": "kehittaminen",
+            "name": {"fin": "Kehittäminen"},
+            "child_values": [
+                "reservialue",
+                "kehitettava",
+                "merkittavastiParannettava",
+                "eheytettavaTaiTiivistettava",
+            ],
+        },
+    ]
 
 
 class TypeOfVerbalPlanRegulation(CodeBase):

database/db_helper.py

@@ -0,0 +1,91 @@
+import enum
+import json
+import os
+from typing import Dict, Tuple
+
+import boto3
+
+
+class User(enum.Enum):
+    SU = "DB_SECRET_SU_ARN"
+    ADMIN = "DB_SECRET_ADMIN_ARN"
+    READ_WRITE = "DB_SECRET_RW_ARN"
+    READ = "DB_SECRET_R_ARN"
+
+
+class Db(enum.Enum):
+    MAINTENANCE = 1
+    MAIN = 2
+
+
+class DatabaseHelper:
+    def __init__(self):
+        if os.environ.get("READ_FROM_AWS", "1") == "1":
+            session = boto3.session.Session()
+            client = session.client(
+                service_name="secretsmanager",
+                region_name=os.environ.get("AWS_REGION_NAME"),
+            )
+            self._users = {
+                user: json.loads(
+                    client.get_secret_value(SecretId=os.environ.get(user.value))[
+                        "SecretString"
+                    ]
+                )
+                for user in User
+            }
+        else:
+            self._users = {
+                User.SU: {
+                    "username": os.environ.get("SU_USER"),
+                    "password": os.environ.get("SU_USER_PW"),
+                },
+                User.ADMIN: {
+                    "username": os.environ.get("ADMIN_USER"),
+                    "password": os.environ.get("ADMIN_USER_PW"),
+                },
+                User.READ_WRITE: {
+                    "username": os.environ.get("RW_USER"),
+                    "password": os.environ.get("RW_USER_PW"),
+                },
+                User.READ: {
+                    "username": os.environ.get("R_USER"),
+                    "password": os.environ.get("R_USER_PW"),
+                },
+            }
+        self._dbs = {
+            Db.MAIN: os.environ.get("DB_MAIN_NAME"),
+            Db.MAINTENANCE: os.environ.get("DB_MAINTENANCE_NAME"),
+        }
+        self._host = os.environ.get("DB_INSTANCE_ADDRESS")
+        self._port = os.environ.get("DB_INSTANCE_PORT", "5432")
+        self._region_name = os.environ.get("AWS_REGION_NAME")
+
+    def get_connection_parameters(
+        self, user: User = User.ADMIN, db: Db = Db.MAIN
+    ) -> Dict[str, str]:
+        user_credentials = self._users.get(user)
+        return {
+            "host": self._host,
+            "port": self._port,
+            "dbname": self.get_db_name(db),
+            "user": user_credentials["username"],
+            "password": user_credentials["password"],
+        }
+
+    def get_connection_string(self) -> str:
+        db_params = self.get_connection_parameters()
+        return (
+            f'postgresql://{db_params["user"]}:{db_params["password"]}'
+            f'@{db_params["host"]}:{db_params["port"]}/{db_params["dbname"]}'
+        )
+
+    def get_username_and_password(self, user: User) -> Tuple[str, str]:
+        user_credentials = self._users.get(user)
+        return user_credentials["username"], user_credentials["password"]
+
+    def get_db_name(self, db: Db) -> str:
+        return self._dbs[db]
+
+    def get_users(self) -> Dict[User, dict]:
+        return self._users

database/db_manager.Dockerfile

@@ -4,6 +4,7 @@ FROM public.ecr.aws/lambda/python:3.12
 COPY database/db_manager/db_manager.py ${LAMBDA_TASK_ROOT}/db_manager.py
 COPY database/migrations ${LAMBDA_TASK_ROOT}/migrations
 COPY database/alembic.ini ${LAMBDA_TASK_ROOT}/alembic.ini
+COPY database/db_helper.py  ${LAMBDA_TASK_ROOT}/db_helper.py
 COPY database/base.py ${LAMBDA_TASK_ROOT}/base.py
 COPY database/codes.py ${LAMBDA_TASK_ROOT}/codes.py
 COPY database/models.py ${LAMBDA_TASK_ROOT}/models.py

database/db_manager/db_manager.py

@@ -1,16 +1,15 @@
 import enum
 import json
 import logging
-import os
 from pathlib import Path
-from typing import Dict, Optional, Tuple, TypedDict
+from typing import Optional, TypedDict
 
-import boto3
 import psycopg2
 from alembic import command
 from alembic.config import Config
 from alembic.script import ScriptDirectory
 from alembic.util.exc import CommandError
+from db_helper import DatabaseHelper, Db, User
 from psycopg2.sql import SQL, Identifier
 
 """
@@ -37,82 +36,6 @@ class Event(TypedDict):
     version: Optional[str]  # Ansible version id
 
 
-class User(enum.Enum):
-    SU = "DB_SECRET_SU_ARN"
-    ADMIN = "DB_SECRET_ADMIN_ARN"
-    READ_WRITE = "DB_SECRET_RW_ARN"
-    READ = "DB_SECRET_R_ARN"
-
-
-class Db(enum.Enum):
-    MAINTENANCE = 1
-    MAIN = 2
-
-
-class DatabaseHelper:
-    def __init__(self):
-        if os.environ.get("READ_FROM_AWS", "1") == "1":
-            session = boto3.session.Session()
-            client = session.client(
-                service_name="secretsmanager",
-                region_name=os.environ.get("AWS_REGION_NAME"),
-            )
-            self._users = {
-                user: json.loads(
-                    client.get_secret_value(SecretId=os.environ.get(user.value))[
-                        "SecretString"
-                    ]
-                )
-                for user in User
-            }
-        else:
-            self._users = {
-                User.SU: {
-                    "username": os.environ.get("SU_USER"),
-                    "password": os.environ.get("SU_USER_PW"),
-                },
-                User.ADMIN: {
-                    "username": os.environ.get("ADMIN_USER"),
-                    "password": os.environ.get("ADMIN_USER_PW"),
-                },
-                User.READ_WRITE: {
-                    "username": os.environ.get("RW_USER"),
-                    "password": os.environ.get("RW_USER_PW"),
-                },
-                User.READ: {
-                    "username": os.environ.get("R_USER"),
-                    "password": os.environ.get("R_USER_PW"),
-                },
-            }
-        self._dbs = {
-            Db.MAIN: os.environ.get("DB_MAIN_NAME"),
-            Db.MAINTENANCE: os.environ.get("DB_MAINTENANCE_NAME"),
-        }
-        self._host = os.environ.get("DB_INSTANCE_ADDRESS")
-        self._port = os.environ.get("DB_INSTANCE_PORT", "5432")
-        self._region_name = os.environ.get("AWS_REGION_NAME")
-
-    def get_connection_parameters(self, user: User, db: Db = Db.MAIN) -> Dict[str, str]:
-        user_credentials = self._users.get(user)
-        return {
-            "host": self._host,
-            "port": self._port,
-            "dbname": self.get_db_name(db),
-            "user": user_credentials["username"],
-            "password": user_credentials["password"],
-        }
-
-    def get_username_and_password(self, user: User) -> Tuple[str, str]:
-        user_credentials = self._users.get(user)
-        return user_credentials["username"], user_credentials["password"]
-
-    def get_db_name(self, db: Db) -> str:
-        return self._dbs[db]
-
-    def get_users(self) -> Dict[User, dict]:
-        return self._users
-
-
 def create_db(conn: psycopg2.extensions.connection, db_name: str) -> str:
     """Creates empty db."""
     with conn.cursor() as cur:

database/koodistot_loader.Dockerfile

@@ -2,6 +2,7 @@ FROM public.ecr.aws/lambda/python:3.12
 
 # Copy function code
 COPY database/koodistot_loader/koodistot_loader.py ${LAMBDA_TASK_ROOT}/koodistot_loader.py
+COPY database/db_helper.py  ${LAMBDA_TASK_ROOT}/db_helper.py
 COPY database/base.py ${LAMBDA_TASK_ROOT}/base.py
 COPY database/codes.py ${LAMBDA_TASK_ROOT}/codes.py
 COPY database/models.py ${LAMBDA_TASK_ROOT}/models.py