61 lines
2.6 KiB
Python
61 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from dataclasses import dataclass, field
|
|
|
|
EXPECTED_PROJECT = "teleo-501523"
|
|
EXPECTED_REGION = "europe-west6"
|
|
EXPECTED_INSTANCE = "teleo-pgvector-standby"
|
|
EXPECTED_CONNECTION_NAME = f"{EXPECTED_PROJECT}:{EXPECTED_REGION}:{EXPECTED_INSTANCE}"
|
|
EXPECTED_DATABASE = "teleo_canonical"
|
|
EXPECTED_AUTHORIZATION_ROLE = "kb_observatory_read"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Settings:
|
|
project_id: str
|
|
instance_connection_name: str
|
|
database: str
|
|
iam_database_user: str
|
|
authorization_role: str
|
|
api_key: str = field(repr=False)
|
|
service_revision: str = "unknown"
|
|
port: int = 8080
|
|
|
|
@classmethod
|
|
def from_env(cls) -> Settings:
|
|
settings = cls(
|
|
project_id=os.environ.get("GCP_PROJECT_ID", ""),
|
|
instance_connection_name=os.environ.get("CLOUD_SQL_INSTANCE", ""),
|
|
database=os.environ.get("DB_NAME", ""),
|
|
iam_database_user=os.environ.get("DB_IAM_USER", ""),
|
|
authorization_role=os.environ.get("DB_AUTHORIZATION_ROLE", ""),
|
|
api_key=os.environ.get("OBSERVATORY_API_KEY", ""),
|
|
service_revision=os.environ.get("SERVICE_REVISION", "unknown"),
|
|
port=int(os.environ.get("PORT", "8080")),
|
|
)
|
|
settings.validate()
|
|
return settings
|
|
|
|
def validate(self) -> None:
|
|
if self.project_id != EXPECTED_PROJECT:
|
|
raise ValueError(f"GCP_PROJECT_ID must be {EXPECTED_PROJECT}")
|
|
if self.instance_connection_name != EXPECTED_CONNECTION_NAME:
|
|
raise ValueError(f"CLOUD_SQL_INSTANCE must be {EXPECTED_CONNECTION_NAME}")
|
|
if self.database != EXPECTED_DATABASE:
|
|
raise ValueError(f"DB_NAME must be {EXPECTED_DATABASE}")
|
|
if self.authorization_role != EXPECTED_AUTHORIZATION_ROLE:
|
|
raise ValueError(f"DB_AUTHORIZATION_ROLE must be {EXPECTED_AUTHORIZATION_ROLE}")
|
|
if not self.iam_database_user.endswith(f"@{EXPECTED_PROJECT}.iam"):
|
|
raise ValueError("DB_IAM_USER must be the scoped Cloud SQL service-account user")
|
|
if any(character.isspace() for character in self.iam_database_user):
|
|
raise ValueError("DB_IAM_USER must not contain whitespace")
|
|
if (
|
|
len(self.api_key) < 32
|
|
or len(self.api_key) > 512
|
|
or not self.api_key.isascii()
|
|
or any(character.isspace() for character in self.api_key)
|
|
):
|
|
raise ValueError("OBSERVATORY_API_KEY must be 32-512 non-whitespace ASCII characters")
|
|
if not 1 <= self.port <= 65535:
|
|
raise ValueError("PORT is outside the valid TCP port range")
|