feat(v9): add creators for TagAttachment, S3Prefix and FlowControlOperation

pyatlan_v9/model/assets/_overlays/flow_control_operation.py

@@ -0,0 +1,77 @@
+# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        connection_qualified_name: str,
+        source_id: str,
+        scope: str | None = None,
+        parent_qualified_name: str | None = None,
+        flow_id: str | None = None,
+    ) -> "FlowControlOperation":
+        """
+        Create a new FlowControlOperation asset.
+
+        Exactly one of ``scope`` or ``parent_qualified_name`` must be given:
+
+        - top-level operation: ``{connection_qualified_name}/{scope}/{source_id}``
+          (e.g. scope = Fabric workspace ID, or SageMaker region)
+        - child operation: ``{parent_qualified_name}/jobs/{source_id}``, linked to
+          its parent through ``flow_controlled_by``
+
+        Args:
+            name: Name of the operation
+            connection_qualified_name: Unique name of the connection
+            source_id: Identifier of the operation in the source system; the last
+                segment of the qualified name
+            scope: Source-system container of a top-level operation
+            parent_qualified_name: Unique name of the controlling operation
+            flow_id: Value for ``flow_id`` (e.g. an ARN); not part of the
+                qualified name
+
+        Returns:
+            FlowControlOperation instance ready to be created
+
+        Raises:
+            ValueError: If required parameters are missing or invalid
+        """
+        validate_required_fields(
+            ["name", "connection_qualified_name", "source_id"],
+            [name, connection_qualified_name, source_id],
+        )
+        fields = connection_qualified_name.split("/")
+        if len(fields) != 3:
+            raise ValueError("Invalid connection_qualified_name")
+        if fields[0].replace(" ", "") == "" or fields[2].replace(" ", "") == "":
+            raise ValueError("Invalid connection_qualified_name")
+        if (scope is None) == (parent_qualified_name is None):
+            raise ValueError(
+                "Exactly one of scope or parent_qualified_name must be provided"
+            )
+
+        flow_controlled_by: RelatedFlowControlOperation | UnsetType = UNSET
+        if parent_qualified_name is not None:
+            validate_required_fields(["parent_qualified_name"], [parent_qualified_name])
+            if not parent_qualified_name.startswith(f"{connection_qualified_name}/"):
+                raise ValueError(
+                    "parent_qualified_name must belong to connection_qualified_name"
+                )
+            qualified_name = f"{parent_qualified_name}/jobs/{source_id}"
+            flow_controlled_by = RelatedFlowControlOperation(
+                qualified_name=parent_qualified_name
+            )
+        else:
+            validate_required_fields(["scope"], [scope])
+            qualified_name = f"{connection_qualified_name}/{scope}/{source_id}"
+
+        return cls(
+            name=name,
+            qualified_name=qualified_name,
+            connection_qualified_name=connection_qualified_name,
+            connector_name=fields[1],
+            flow_id=flow_id if flow_id is not None else UNSET,
+            flow_controlled_by=flow_controlled_by,
+        )

pyatlan_v9/model/assets/_overlays/s3_prefix.py

@@ -0,0 +1,89 @@
+# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        connection_qualified_name: str,
+        s3_bucket_name: str,
+        s3_bucket_qualified_name: str,
+        prefix: str,
+        name: str | None = None,
+    ) -> "S3Prefix":
+        """
+        Create a new S3Prefix asset.
+
+        The qualified name is ``{s3_bucket_qualified_name}/{prefix}/`` — the full
+        key from the bucket root, always with a trailing slash (native S3
+        convention, shared by every app that builds prefixes).
+
+        The parent prefix is recorded by qualified name
+        (``s3_parent_prefix_qualified_name`` and ``s3_prefix_hierarchy``); the
+        ``s3_parent_prefix`` relationship is left for the caller to set once the
+        parent prefix exists.
+
+        Args:
+            connection_qualified_name: Unique name of the connection
+            s3_bucket_name: Simple name of the bucket
+            s3_bucket_qualified_name: Unique name of the bucket
+            prefix: Full key of the prefix from the bucket root, e.g. ``a/b/``
+                (leading and trailing slashes are optional)
+            name: Name of the prefix; defaults to the last segment of ``prefix``
+
+        Returns:
+            S3Prefix instance ready to be created
+
+        Raises:
+            ValueError: If required parameters are missing or invalid
+        """
+        validate_required_fields(
+            [
+                "connection_qualified_name",
+                "s3_bucket_name",
+                "s3_bucket_qualified_name",
+                "prefix",
+            ],
+            [
+                connection_qualified_name,
+                s3_bucket_name,
+                s3_bucket_qualified_name,
+                prefix,
+            ],
+        )
+        fields = connection_qualified_name.split("/")
+        if len(fields) != 3:
+            raise ValueError("Invalid connection_qualified_name")
+        if fields[0].replace(" ", "") == "" or fields[2].replace(" ", "") == "":
+            raise ValueError("Invalid connection_qualified_name")
+        if fields[1].lower() != "s3":
+            raise ValueError("Invalid connection_qualified_name")
+
+        segments = prefix.strip("/").split("/")
+        if any(segment.strip() == "" for segment in segments):
+            raise ValueError("prefix cannot contain empty segments")
+        if name is None:
+            name = segments[-1]
+        validate_required_fields(["name"], [name])
+
+        def _prefix_qn(depth: int) -> str:
+            return f"{s3_bucket_qualified_name}/{'/'.join(segments[:depth])}/"
+
+        # Ancestors from immediate parent up to the root prefix.
+        hierarchy = [
+            {"name": segments[depth - 1], "qualifiedName": _prefix_qn(depth)}
+            for depth in range(len(segments) - 1, 0, -1)
+        ]
+        return cls(
+            name=name,
+            qualified_name=_prefix_qn(len(segments)),
+            connection_qualified_name=connection_qualified_name,
+            connector_name=fields[1],
+            s3_bucket_name=s3_bucket_name,
+            s3_bucket_qualified_name=s3_bucket_qualified_name,
+            s3_bucket=RelatedS3Bucket(qualified_name=s3_bucket_qualified_name),
+            s3_parent_prefix_qualified_name=(
+                hierarchy[0]["qualifiedName"] if hierarchy else UNSET
+            ),
+            s3_prefix_hierarchy=hierarchy or UNSET,
+        )

pyatlan_v9/model/assets/_overlays/tag_attachment.py

@@ -0,0 +1,98 @@
+# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields
+
+    _QN_STYLES: ClassVar[tuple[str, ...]] = (
+        "tag_object",
+        "tag_attachments_object",
+        "object_tag",
+    )
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        connection_qualified_name: str,
+        tag_qualified_name: str,
+        object_qualified_name: str,
+        qn_style: str = "tag_object",
+        object_type_name: str | None = None,
+        source_tag_type_name: str | None = None,
+        value: str | None = None,
+    ) -> "TagAttachment":
+        """
+        Create a new TagAttachment asset, linking a source tag to an asset.
+
+        Apps already in production build the qualified name in different shapes,
+        so ``qn_style`` selects one; changing it changes the attachment's identity:
+
+        - ``tag_object`` (default): ``{tag_qualified_name}/{object_qualified_name}``
+        - ``tag_attachments_object``:
+          ``{tag_qualified_name}/attachments/{object_qualified_name}``
+        - ``object_tag``: ``{object_qualified_name}/{name}``
+
+        Args:
+            name: Name of the source tag
+            connection_qualified_name: Unique name of the connection
+            tag_qualified_name: Unique name of the source tag
+            object_qualified_name: Unique name of the asset the tag is attached to
+            qn_style: Shape of the qualified name (see above)
+            object_type_name: Type name of the asset the tag is attached to
+            source_tag_type_name: Type name of the source tag
+            value: String value of the tag on this asset, if any
+
+        Returns:
+            TagAttachment instance ready to be created
+
+        Raises:
+            ValueError: If required parameters are missing or invalid
+        """
+        validate_required_fields(
+            [
+                "name",
+                "connection_qualified_name",
+                "tag_qualified_name",
+                "object_qualified_name",
+            ],
+            [
+                name,
+                connection_qualified_name,
+                tag_qualified_name,
+                object_qualified_name,
+            ],
+        )
+        fields = connection_qualified_name.split("/")
+        if len(fields) != 3:
+            raise ValueError("Invalid connection_qualified_name")
+        if fields[0].replace(" ", "") == "" or fields[2].replace(" ", "") == "":
+            raise ValueError("Invalid connection_qualified_name")
+
+        if qn_style == "tag_object":
+            qualified_name = f"{tag_qualified_name}/{object_qualified_name}"
+        elif qn_style == "tag_attachments_object":
+            qualified_name = f"{tag_qualified_name}/attachments/{object_qualified_name}"
+        elif qn_style == "object_tag":
+            qualified_name = f"{object_qualified_name}/{name}"
+        else:
+            raise ValueError(f"qn_style must be one of {cls._QN_STYLES}")
+
+        return cls(
+            name=name,
+            qualified_name=qualified_name,
+            connection_qualified_name=connection_qualified_name,
+            connector_name=fields[1],
+            tag_qualified_name=tag_qualified_name,
+            source_tag_qualified_name=tag_qualified_name,
+            source_tag_display_name=name,
+            source_tag_type_name=(
+                source_tag_type_name if source_tag_type_name is not None else UNSET
+            ),
+            object_qualified_name=object_qualified_name,
+            object_type_name=object_type_name
+            if object_type_name is not None
+            else UNSET,
+            value_type="NONE" if value is None else "STRING",
+            value=value,
+            tag_attachment_string_value=value if value is not None else UNSET,
+            propagate=False,
+        )

pyatlan_v9/model/assets/flow_control_operation.py

@@ -25,6 +25,7 @@
 )
 from pyatlan_v9.model.serde import Serde, get_serde
 from pyatlan_v9.model.transform import register_asset
+from pyatlan_v9.utils import init_guid, validate_required_fields
 
 from .anomalo_related import RelatedAnomaloCheck
 from .app_related import RelatedApplication, RelatedApplicationField
@@ -315,6 +316,82 @@ def relate(self) -> "RelatedFlowControlOperation":
             return RelatedFlowControlOperation(guid=self.guid)
         return RelatedFlowControlOperation(qualified_name=self.qualified_name)
 
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        connection_qualified_name: str,
+        source_id: str,
+        scope: str | None = None,
+        parent_qualified_name: str | None = None,
+        flow_id: str | None = None,
+    ) -> "FlowControlOperation":
+        """
+        Create a new FlowControlOperation asset.
+
+        Exactly one of ``scope`` or ``parent_qualified_name`` must be given:
+
+        - top-level operation: ``{connection_qualified_name}/{scope}/{source_id}``
+          (e.g. scope = Fabric workspace ID, or SageMaker region)
+        - child operation: ``{parent_qualified_name}/jobs/{source_id}``, linked to
+          its parent through ``flow_controlled_by``
+
+        Args:
+            name: Name of the operation
+            connection_qualified_name: Unique name of the connection
+            source_id: Identifier of the operation in the source system; the last
+                segment of the qualified name
+            scope: Source-system container of a top-level operation
+            parent_qualified_name: Unique name of the controlling operation
+            flow_id: Value for ``flow_id`` (e.g. an ARN); not part of the
+                qualified name
+
+        Returns:
+            FlowControlOperation instance ready to be created
+
+        Raises:
+            ValueError: If required parameters are missing or invalid
+        """
+        validate_required_fields(
+            ["name", "connection_qualified_name", "source_id"],
+            [name, connection_qualified_name, source_id],
+        )
+        fields = connection_qualified_name.split("/")
+        if len(fields) != 3:
+            raise ValueError("Invalid connection_qualified_name")
+        if fields[0].replace(" ", "") == "" or fields[2].replace(" ", "") == "":
+            raise ValueError("Invalid connection_qualified_name")
+        if (scope is None) == (parent_qualified_name is None):
+            raise ValueError(
+                "Exactly one of scope or parent_qualified_name must be provided"
+            )
+
+        flow_controlled_by: RelatedFlowControlOperation | UnsetType = UNSET
+        if parent_qualified_name is not None:
+            validate_required_fields(["parent_qualified_name"], [parent_qualified_name])
+            if not parent_qualified_name.startswith(f"{connection_qualified_name}/"):
+                raise ValueError(
+                    "parent_qualified_name must belong to connection_qualified_name"
+                )
+            qualified_name = f"{parent_qualified_name}/jobs/{source_id}"
+            flow_controlled_by = RelatedFlowControlOperation(
+                qualified_name=parent_qualified_name
+            )
+        else:
+            validate_required_fields(["scope"], [scope])
+            qualified_name = f"{connection_qualified_name}/{scope}/{source_id}"
+
+        return cls(
+            name=name,
+            qualified_name=qualified_name,
+            connection_qualified_name=connection_qualified_name,
+            connector_name=fields[1],
+            flow_id=flow_id if flow_id is not None else UNSET,
+            flow_controlled_by=flow_controlled_by,
+        )
+
     # =========================================================================
     # Optimized Serialization Methods (override Asset base class)
     # =========================================================================