refactor: revert refactor: Refactoring VPC provider" (#706)

dbt_platform_helper/domain/database_copy.py

@@ -11,12 +11,12 @@
 from dbt_platform_helper.domain.maintenance_page import MaintenancePageProvider
 from dbt_platform_helper.providers.aws import AWSException
 from dbt_platform_helper.providers.config import ConfigProvider
-from dbt_platform_helper.providers.vpc import Vpc
-from dbt_platform_helper.providers.vpc import VpcProvider
 from dbt_platform_helper.utils.application import Application
 from dbt_platform_helper.utils.application import ApplicationNotFoundException
 from dbt_platform_helper.utils.application import load_application
+from dbt_platform_helper.utils.aws import Vpc
 from dbt_platform_helper.utils.aws import get_connection_string
+from dbt_platform_helper.utils.aws import get_vpc_info_by_name
 from dbt_platform_helper.utils.aws import wait_for_log_group_to_exist
 from dbt_platform_helper.utils.messages import abort_with_error
 
@@ -28,7 +28,7 @@ def __init__(
         database: str,
         auto_approve: bool = False,
         load_application: Callable[[str], Application] = load_application,
-        vpc_config: Callable[[Session, str, str, str], Vpc] = VpcProvider.get_vpc_info_by_name,
+        vpc_config: Callable[[Session, str, str, str], Vpc] = get_vpc_info_by_name,
         db_connection_string: Callable[
             [Session, str, str, str, Callable], str
         ] = get_connection_string,

dbt_platform_helper/utils/application.py

@@ -1,8 +1,6 @@
 import json
 import os
 import re
-from dataclasses import dataclass
-from dataclasses import field
 from pathlib import Path
 from typing import Dict
 
@@ -18,12 +16,16 @@
 from dbt_platform_helper.utils.messages import abort_with_error
 
 
-@dataclass
 class Environment:
     name: str
     account_id: str
     sessions: Dict[str, boto3.Session]
 
+    def __init__(self, name: str, account_id: str, sessions: Dict[str, boto3.Session]):
+        self.name = name
+        self.account_id = account_id
+        self.sessions = sessions
+
     @property
     def session(self):
         if self.account_id not in self.sessions:
@@ -34,17 +36,24 @@ def session(self):
         return self.sessions[self.account_id]
 
 
-@dataclass
 class Service:
     name: str
     kind: str
 
+    def __init__(self, name: str, kind: str):
+        self.name = name
+        self.kind = kind
+
 
-@dataclass
 class Application:
     name: str
-    environments: Dict[str, Environment] = field(default_factory=dict)
-    services: Dict[str, Service] = field(default_factory=dict)
+    environments: Dict[str, Environment]
+    services: Dict[str, Service]
+
+    def __init__(self, name: str):
+        self.name = name
+        self.environments = {}
+        self.services = {}
 
     def __str__(self):
         output = f"Application {self.name} with"

dbt_platform_helper/utils/aws.py

@@ -15,6 +15,7 @@
 
 from dbt_platform_helper.constants import REFRESH_TOKEN_MESSAGE
 from dbt_platform_helper.platform_exception import PlatformException
+from dbt_platform_helper.providers.aws import AWSException
 from dbt_platform_helper.providers.aws import CopilotCodebaseNotFoundException
 from dbt_platform_helper.providers.aws import ImageNotFoundException
 from dbt_platform_helper.providers.aws import LogGroupNotFoundException
@@ -376,6 +377,55 @@ def get_connection_string(
     return f"postgres://{conn['username']}:{conn['password']}@{conn['host']}:{conn['port']}/{conn['dbname']}"
 
 
+class Vpc:
+    def __init__(self, subnets: list[str], security_groups: list[str]):
+        self.subnets = subnets
+        self.security_groups = security_groups
+
+
+def get_vpc_info_by_name(session: Session, app: str, env: str, vpc_name: str) -> Vpc:
+    ec2_client = session.client("ec2")
+    vpc_response = ec2_client.describe_vpcs(Filters=[{"Name": "tag:Name", "Values": [vpc_name]}])
+
+    matching_vpcs = vpc_response.get("Vpcs", [])
+
+    if not matching_vpcs:
+        raise AWSException(f"VPC not found for name '{vpc_name}'")
+
+    vpc_id = vpc_response["Vpcs"][0].get("VpcId")
+
+    if not vpc_id:
+        raise AWSException(f"VPC id not present in vpc '{vpc_name}'")
+
+    ec2_resource = session.resource("ec2")
+    vpc = ec2_resource.Vpc(vpc_id)
+
+    route_tables = ec2_client.describe_route_tables(
+        Filters=[{"Name": "vpc-id", "Values": [vpc_id]}]
+    )["RouteTables"]
+
+    subnets = []
+    for route_table in route_tables:
+        private_routes = [route for route in route_table["Routes"] if "NatGatewayId" in route]
+        if not private_routes:
+            continue
+        for association in route_table["Associations"]:
+            if "SubnetId" in association:
+                subnet_id = association["SubnetId"]
+                subnets.append(subnet_id)
+
+    if not subnets:
+        raise AWSException(f"No private subnets found in vpc '{vpc_name}'")
+
+    tag_value = {"Key": "Name", "Value": f"copilot-{app}-{env}-env"}
+    sec_groups = [sg.id for sg in vpc.security_groups.all() if sg.tags and tag_value in sg.tags]
+
+    if not sec_groups:
+        raise AWSException(f"No matching security groups found in vpc '{vpc_name}'")
+
+    return Vpc(subnets, sec_groups)
+
+
 def start_build_extraction(codebuild_client, build_options):
     response = codebuild_client.start_build(**build_options)
     return response["build"]["arn"]

tests/platform_helper/domain/test_database_copy.py

@@ -8,9 +8,9 @@
 from dbt_platform_helper.domain.database_copy import DatabaseCopy
 from dbt_platform_helper.providers.aws import AWSException
 from dbt_platform_helper.providers.config import ConfigProvider
-from dbt_platform_helper.providers.vpc import Vpc
 from dbt_platform_helper.utils.application import Application
 from dbt_platform_helper.utils.application import ApplicationNotFoundException
+from dbt_platform_helper.utils.aws import Vpc
 
 
 class DataCopyMocks: