from typing import Literal from cryptography.fernet import Fernet from django.conf import settings from labs.utils.database_fields import get_legacy_fernet_key class TiDBServerlessClusterConfig: environment: Literal["user_hosted", "sandbox"] cluster_name: str cluster_id: str cluster_host: str cluster_port: int admin_name: str admin_password: str database_name: str organization_id: str project_id: str def __init__( self, environment: Literal["user_hosted", "sandbox"], cluster_id: str, cluster_name: str, cluster_host: str, cluster_port: int, admin_name: str, admin_password: str, database_name: str = "", organization_id: str = "", project_id: str = "", ): self.fernet = Fernet(get_legacy_fernet_key(settings.SECRET_KEY)) self.environment = environment self.cluster_id = cluster_id self.cluster_name = cluster_name self.cluster_host = cluster_host self.cluster_port = cluster_port self.database_name = database_name self.admin_name = admin_name self.admin_password = admin_password self.organization_id = organization_id self.project_id = project_id def _encrypt(self, value: str) -> str: return self.fernet.encrypt(value.encode("utf-8")).decode("utf-8") def _decrypt(self, value: str) -> str: return self.fernet.decrypt(value.encode("utf-8")).decode("utf-8") @classmethod def from_encrypted_dict(cls, data: dict) -> "TiDBServerlessClusterConfig": instance = cls( environment=data["environment"], cluster_id=data["cluster_id"], cluster_name=data["cluster_name"], cluster_host=data["cluster_host"], cluster_port=data["cluster_port"], admin_name="", admin_password="", database_name=data["database_name"], organization_id=data["organization_id"], project_id=data["project_id"], ) instance.admin_name = instance._decrypt(data["admin_name"]) instance.admin_password = instance._decrypt(data["admin_password"]) return instance def to_encrypted_dict(self) -> dict: return { "environment": self.environment, "cluster_id": self.cluster_id, "cluster_name": self.cluster_name, "cluster_host": self.cluster_host, "cluster_port": self.cluster_port, "database_name": self.database_name, "admin_name": self._encrypt(self.admin_name), "admin_password": self._encrypt(self.admin_password), "organization_id": self.organization_id, "project_id": self.project_id, }