from enum import Enum from typing import Dict, List, Optional, Set from datadog_api_client.v2.model.entity_data import EntityData from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, field_validator class AwsServiceEnum(str, Enum): """ Enum for values of the `aws_service` tag. """ fargate = "fargate" lambda_aws = "lambda" class ServiceTypeEnum(str, Enum): """ Enum for allowed service types. """ web = "web" graphql = "graphql" function = "function" task = "task" daemon = "daemon" connector = "connector" class LibraryTypeEnum(str, Enum): """ Enum for allowed library types. """ sdk = "sdk" framework = "framework" utility = "utility" class FrontendTypeEnum(str, Enum): """ Enum for allowed frontend types. """ web = "web" mobile = "mobile" browser = "browser" class Tags(BaseModel): """ Represents the tags for a service in the Datadog Software Catalog. """ model_config = ConfigDict(extra="allow") # Allow arbitrary fields application_family: Optional[str] = None aws_service: Optional[AwsServiceEnum] = None public: Optional[bool] = None service_type: Optional[ServiceTypeEnum] = None terraformed: Optional[bool] = None @classmethod def to_dict(cls, tags: List[str]) -> Dict[str, str]: """ Converts a list of tags in the format 'key:value' to a dictionary. :param tags: List of tags in the format 'key:value'. :return: Dictionary with keys as tag names and values as tag values. """ if not tags: return {} return {tag.split(":")[0]: tag.split(":")[1] for tag in tags if ":" in tag} class ApiVersionEnum(str, Enum): """ Enum for API versions in the Datadog Software Catalog. """ v3 = "v3" class KindEnum(str, Enum): """ Enum for kinds of entities in the Datadog Software Catalog. """ service = "service" library = "library" frontend = "frontend" system = "system" class MetadataAdditionalOwnersItem(BaseModel): """ Represents additional owners for a service in the Datadog Software Catalog. """ name: str type: Optional[str] = None class MetadataContactsItem(BaseModel): """ Represents contacts for a service in the Datadog Software Catalog. """ contact: str type: str name: Optional[str] = None class MetadataLinksItem(BaseModel): """ Represents links for a service in the Datadog Software Catalog. """ name: str type: str url: str provider: Optional[str] = None class EntityMetadata(BaseModel): """ Base class for metadata in the Datadog Software Catalog. """ name: str description: str owner: Optional[str] = Field(default=None, validate_default=True) tags: Optional[list[str]] = Field(default=[], validate_default=True) additionalOwners: Optional[list[MetadataAdditionalOwnersItem]] = [] contacts: Optional[list[MetadataContactsItem]] = [] displayName: Optional[str] = None inheritFrom: Optional[str] = None links: Optional[list[MetadataLinksItem]] = [] @field_validator("owner", mode="after") def _validate_owner(cls, owner: str | None, info: ValidationInfo) -> str: if isinstance(info.context, dict): # If owner is not specified, check if it is inherited from a system entity system: EntityData = info.context.get("system") if not owner and system: owner = system.attributes.owner if not owner: raise ValueError("owner must be specified or inherited from a system entity.") allowed_teams: Set[str] = info.context.get("teams", set()) if owner not in allowed_teams: raise ValueError( f"Owner '{owner}' is not the handle of an existing Datadog Team. Teams: {allowed_teams}" ) return owner @field_validator("tags", mode="after") def _validate_tags(cls, tags: List[str], info: ValidationInfo) -> List[str]: tags_dict = Tags.to_dict(tags) Tags.model_validate(tags_dict) # Perform additional validation if context is provided if isinstance(info.context, dict): # If the service inherits metadata from a system entity, validate the combined set of tags system: EntityData = info.context.get("system") combined_tags = tags_dict | Tags.to_dict(system.attributes.tags or []) if system else tags_dict missing_tags = cls.required_tag_set() - set(combined_tags.keys()) if missing_tags: raise ValueError( f"Required tags must be specified or inherited from a system entity: " f"{', '.join(sorted(missing_tags))}" ) return tags @classmethod def required_tag_set(cls) -> Set[str]: """ Sub-classes override this to declare tags that *must* be present. The base class declares no mandatory tags – suitable for libraries. """ return set() def get_tags(self) -> Tags: """Returns the tags as a Tags model object.""" return Tags(**Tags.to_dict(self.tags)) class ServiceMetadata(EntityMetadata): """ Represents metadata for a service in the Datadog Software Catalog. """ @classmethod def required_tag_set(cls) -> Set[str]: # Services must expose a public tag (or inherit it from a system entity). return {"public"} class LibraryMetadata(EntityMetadata): """ Represents metadata for a library in the Datadog Software Catalog. """ pass class FrontendMetadata(EntityMetadata): """ Represents metadata for a frontend in the Datadog Software Catalog. """ pass class DatadogCodeLocationItem(BaseModel): """ Represents a code location for a service in the Datadog Software Catalog. """ repositoryURL: str paths: Optional[list[str]] = [] class DatadogEventItem(BaseModel): """ Represents an events association for a service in the Datadog Software Catalog. """ name: Optional[str] = None query: Optional[str] = None class DatadogLogItem(BaseModel): """ Represents a logs association for a service in the Datadog Software Catalog. """ name: Optional[str] = None query: Optional[str] = None class DatadogPerformanceData(BaseModel): """ Represents performance data for a service in the Datadog Software Catalog. """ tags: Optional[list[str]] = [] class DatadogPipelines(BaseModel): """ Represents CI pipelines association for a service in the Datadog Software Catalog. """ fingerprints: Optional[list[str]] = [] class Datadog(BaseModel): """ Represents Datadog product integrations for the service entity. """ codeLocations: Optional[list[DatadogCodeLocationItem]] = [] events: Optional[list[DatadogEventItem]] = [] logs: Optional[list[DatadogLogItem]] = [] performanceData: Optional[DatadogPerformanceData] = None pipelines: Optional[DatadogPipelines] = None class IntegrationsPagerduty(BaseModel): """ Represents PagerDuty integration for a service in the Datadog Software Catalog. """ serviceURL: str class Integrations(BaseModel): """ Represents integrations for a service in the Datadog Software Catalog. """ pagerduty: Optional[IntegrationsPagerduty] = None class BaseSpecModel(BaseModel): """ Base class for specifications in the Datadog Software Catalog. """ model_config = ConfigDict(use_enum_values=True) componentOf: Optional[list[str]] = [] dependsOn: Optional[list[str]] = [] languages: Optional[set[str]] = set() lifecycle: Optional[str] = None tier: Optional[str] = None class ServiceSpec(BaseSpecModel): """ Represents the specification for a service in the Datadog Software Catalog. """ type: ServiceTypeEnum class LibrarySpec(BaseSpecModel): """ Represents the specification for a library in the Datadog Software Catalog. """ type: LibraryTypeEnum class FrontendSpec(BaseSpecModel): """ Represents the specification for a frontend in the Datadog Software Catalog. """ type: FrontendTypeEnum class BaseMetadataSourceExtension(BaseModel): """ Extension for storing information about the source of the service metadata. """ repositoryURL: Optional[str] = None repositoryPath: Optional[str] = None pipelineURL: Optional[str] = None revision: Optional[str] = None class ServiceMetadataSourceExtension(BaseMetadataSourceExtension): """ Represents the metadata source extension for a service in the Datadog Software Catalog. """ serviceDefinitionFilePath: Optional[str] = None servicePath: Optional[str] = None class LibraryMetadataSourceExtension(BaseMetadataSourceExtension): """ Represents the metadata source extension for a library in the Datadog Software Catalog. """ libraryDefinitionFilePath: Optional[str] = None libraryPath: Optional[str] = None class FrontendMetadataSourceExtension(BaseMetadataSourceExtension): """ Represents the metadata source extension for a frontend in the Datadog Software Catalog. """ frontendDefinitionFilePath: Optional[str] = None frontendPath: Optional[str] = None class ServiceExtensions(BaseModel): """ Represents extensions for a service in the Datadog Software Catalog. """ metadata_source: ServiceMetadataSourceExtension = Field( alias="sonymusic-pde.com/metadata-source", default=ServiceMetadataSourceExtension(), ) class LibraryExtensions(BaseModel): """ Represents extensions for a library in the Datadog Software Catalog. """ metadata_source: LibraryMetadataSourceExtension = Field( alias="sonymusic-pde.com/metadata-source", default=LibraryMetadataSourceExtension(), ) class FrontendExtensions(BaseModel): """ Represents extensions for a frontend in the Datadog Software Catalog. """ metadata_source: FrontendMetadataSourceExtension = Field( alias="sonymusic-pde.com/metadata-source", default=FrontendMetadataSourceExtension(), ) class BaseEntityModel(BaseModel): """ Base class for entities in the Datadog Software Catalog. """ model_config = ConfigDict(use_enum_values=True) apiVersion: ApiVersionEnum kind: KindEnum datadog: Optional[Datadog] = Datadog() class Service(BaseEntityModel): """ Represents a service in the Datadog Software Catalog. """ extensions: Optional[ServiceExtensions] = ServiceExtensions() spec: ServiceSpec integrations: Optional[Integrations] = Integrations() metadata: ServiceMetadata class Library(BaseEntityModel): """ Represents a library in the Datadog Software Catalog. """ metadata: LibraryMetadata extensions: Optional[LibraryExtensions] = LibraryExtensions() spec: LibrarySpec class Frontend(BaseEntityModel): """ Represents a frontend in the Datadog Software Catalog. """ metadata: FrontendMetadata extensions: Optional[FrontendExtensions] = FrontendExtensions() spec: FrontendSpec