class Primitive:
    primitive_type: PrimitiveType
    _managed: bool = False

    def primitive_id(self):
        raise NotImplementedError(
            f"Primitive.primitive_id() is not implemented for type: {type(self)}."
        )

    def enable_managed(self):
        self._managed = True

    def options(self) -> "Primitive":
        return self

    def pulumi_resources_if_managed(
        self,
        credentials: CredentialType,
        opts: pulumi.ResourceOptions,
    ):
        if not self._managed:
            return None
        return self.pulumi_resources(credentials, opts)

    def pulumi_resources(
        self,
        credentials: CredentialType,
        opts: pulumi.ResourceOptions,
    ) -> List[pulumi.Resource]:
        raise NotImplementedError(
            f"Primitive.pulumi_resources() is not implemented for type: {type(self)}."
        )

    def source(self, credentials: CredentialType) -> SourceStrategy:
        raise NotImplementedError(
            f"Primitive.source() is not implemented for type: {type(self)}."
        )

    def sink(self, credentials: CredentialType) -> SinkStrategy:
        raise NotImplementedError(
            f"Primitive.sink() is not implemented for type: {type(self)}."
        )

    def background_tasks(self, credentials: CredentialType) -> List[BackgroundTask]:
        return []

    def dependency(self) -> "PrimitiveDependency":
        """Returns a dependency allowing the primitive to be injected as a dependency.

        For example:

        prim = Primitive(...)
        PrimDep = prim.dependency()

        @dependency(scope=Scope.PROCESS)
        class MyDep:
        def __init__(self, injected_primitive: PrimDep):
            self.prim_field = injected_primitive.field_on_primitive
        """
        return PrimitiveDependency(self)

    def cloud_console_url(self) -> Optional[str]:
        """Returns a URL to the cloud console for this primitive."""
        return Nonemanaged
        return self

​
GCP Primitive

class GCPPrimtive(Primitive):
    primitive_type = PrimitiveType.GCP

    @classmethod
    def from_gcp_options(cls, gcp_options: GCPOptions) -> "GCPPrimtive":
        """Create a primitive from GCPOptions."""
        raise NotImplementedError("GCPPrimtive.from_gcp_options() is not implemented.")

​
AWS Primitive

class AWSPrimtive(Primitive):
    primitive_type = PrimitiveType.AWS

    @classmethod
    def from_aws_options(cls, aws_options: AWSOptions) -> "AWSPrimtive":
        """Create a primitive from AWSOptions."""
        raise NotImplementedError("AWSPrimtive.from_aws_options() is not implemented.")
CredentialsStrategy