Initial version

Author: ErnestoLoma

pyatlan/model/assets/core/procedure.py

@@ -4,18 +4,68 @@
 
 from __future__ import annotations
 
-from typing import ClassVar, List, Optional
+from typing import ClassVar, List, Optional, overload
 
 from pydantic.v1 import Field, validator
 
 from pyatlan.model.fields.atlan_fields import RelationField, TextField
+from pyatlan.utils import init_guid, validate_required_fields
 
+from ...enums import AtlanConnectorType
 from .s_q_l import SQL
 
 
 class Procedure(SQL):
     """Description"""
 
+    @overload
+    @classmethod
+    def creator(
+        cls,
+        *,
+        name: str,
+        schema_qualified_name: str,
+        definition: str,
+    ) -> Procedure: ...
+
+    @overload
+    @classmethod
+    def creator(
+        cls,
+        *,
+        name: str,
+        schema_qualified_name: str,
+        schema_name: str,
+        database_name: str,
+        database_qualified_name: str,
+        connection_qualified_name: str,
+        definition: str,
+    ) -> Procedure: ...
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        definition: str,
+        schema_qualified_name: str,
+        schema_name: Optional[str] = None,
+        database_name: Optional[str] = None,
+        database_qualified_name: Optional[str] = None,
+        connection_qualified_name: Optional[str] = None,
+    ) -> Procedure:
+        attributes = Procedure.Attributes.create(
+            name=name,
+            definition=definition,
+            schema_qualified_name=schema_qualified_name,
+            schema_name=schema_name,
+            database_name=database_name,
+            database_qualified_name=database_qualified_name,
+            connection_qualified_name=connection_qualified_name,
+        )
+        return cls(attributes=attributes)
+
     type_name: str = Field(default="Procedure", allow_mutation=False)
 
     @validator("type_name")
@@ -70,6 +120,56 @@ class Attributes(SQL.Attributes):
             default=None, description=""
         )  # relationship
 
+        @classmethod
+        @init_guid
+        def create(
+            cls,
+            *,
+            name: str,
+            definition: str,
+            schema_qualified_name: Optional[str] = None,
+            schema_name: Optional[str] = None,
+            database_name: Optional[str] = None,
+            database_qualified_name: Optional[str] = None,
+            connection_qualified_name: Optional[str] = None,
+        ) -> Procedure.Attributes:
+            validate_required_fields(
+                ["name", "definition", "schema_qualified_name"],
+                [name, definition, schema_qualified_name],
+            )
+            assert schema_qualified_name  # noqa: S101
+            if connection_qualified_name:
+                connector_name = AtlanConnectorType.get_connector_name(
+                    connection_qualified_name
+                )
+            else:
+                connection_qn, connector_name = AtlanConnectorType.get_connector_name(
+                    schema_qualified_name, "schema_qualified_name", 5
+                )
+
+            fields = schema_qualified_name.split("/")
+            qualified_name = f"{schema_qualified_name}/_procedures_/{name}"
+            connection_qualified_name = connection_qualified_name or connection_qn
+            database_name = database_name or fields[3]
+            schema_name = schema_name or fields[4]
+            database_qualified_name = (
+                database_qualified_name
+                or f"{connection_qualified_name}/{database_name}"
+            )
+
+            return Procedure.Attributes(
+                name=name,
+                definition=definition,
+                qualified_name=qualified_name,
+                database_name=database_name,
+                database_qualified_name=database_qualified_name,
+                schema_name=schema_name,
+                schema_qualified_name=schema_qualified_name,
+                atlan_schema=Schema.ref_by_qualified_name(schema_qualified_name),
+                connector_name=connector_name,
+                connection_qualified_name=connection_qualified_name,
+            )
+
     attributes: Procedure.Attributes = Field(
         default_factory=lambda: Procedure.Attributes(),
         description=(

tests/unit/model/constants.py

@@ -237,3 +237,4 @@
 SUPERSET_DATASET_QUALIFIED_NAME = (
     f"{SUPERSET_DASHBOARD_QUALIFIED_NAME}/{SUPERSET_DATASET_NAME}"
 )
+PROCEDURE_NAME = "test-procedure"

tests/unit/model/procedure_test.py

@@ -0,0 +1,82 @@
+import pytest
+
+from pyatlan.model.assets import Procedure
+from tests.unit.model.constants import (
+    CONNECTION_QUALIFIED_NAME,
+    CONNECTOR_TYPE,
+    DATABASE_NAME,
+    DATABASE_QUALIFIED_NAME,
+    PROCEDURE_NAME,
+    SCHEMA_NAME,
+    SCHEMA_QUALIFIED_NAME,
+)
+
+DEFINITION = """
+BEGIN
+insert into `atlanhq.testing_lineage.INSTACART_ALCOHOL_ORDER_TIME_copy`
+select * from `atlanhq.testing_lineage.INSTACART_ALCOHOL_ORDER_TIME`;
+END
+"""
+
+
+@pytest.mark.parametrize(
+    "name, definition, schema_qualified_name, message",
+    [
+        (None, DEFINITION, SCHEMA_QUALIFIED_NAME, "name is required"),
+        (PROCEDURE_NAME, None, SCHEMA_QUALIFIED_NAME, "definition is required"),
+        (PROCEDURE_NAME, DEFINITION, None, "schema_qualified_name is required"),
+    ],
+)
+def test_create_with_missing_parameters_raise_value_error(
+    name: str, definition: str, schema_qualified_name: str, message: str
+):
+    with pytest.raises(ValueError, match=message):
+        Procedure.create(
+            name=name,
+            definition=definition,
+            schema_qualified_name=schema_qualified_name,
+        )
+
+
+def test_creator():
+    sut = Procedure.create(
+        name=PROCEDURE_NAME,
+        definition=DEFINITION,
+        schema_qualified_name=SCHEMA_QUALIFIED_NAME,
+    )
+
+    assert sut.name == PROCEDURE_NAME
+    assert sut.database_name == DATABASE_NAME
+    assert sut.connection_qualified_name == CONNECTION_QUALIFIED_NAME
+    assert sut.database_qualified_name == DATABASE_QUALIFIED_NAME
+    assert (
+        sut.qualified_name == f"{SCHEMA_QUALIFIED_NAME}/_procedures_/{PROCEDURE_NAME}"
+    )
+    assert sut.schema_qualified_name == SCHEMA_QUALIFIED_NAME
+    assert sut.schema_name == SCHEMA_NAME
+    assert sut.connector_name == CONNECTOR_TYPE
+    assert sut.atlan_schema.qualified_name == SCHEMA_QUALIFIED_NAME
+
+
+def test_overload_creator():
+    sut = Procedure.creator(
+        name=PROCEDURE_NAME,
+        definition=DEFINITION,
+        schema_qualified_name=SCHEMA_QUALIFIED_NAME,
+        schema_name=SCHEMA_NAME,
+        database_name=DATABASE_NAME,
+        database_qualified_name=DATABASE_QUALIFIED_NAME,
+        connection_qualified_name=CONNECTION_QUALIFIED_NAME,
+    )
+
+    assert sut.name == PROCEDURE_NAME
+    assert sut.database_name == DATABASE_NAME
+    assert sut.connection_qualified_name == CONNECTION_QUALIFIED_NAME
+    assert sut.database_qualified_name == DATABASE_QUALIFIED_NAME
+    assert (
+        sut.qualified_name == f"{SCHEMA_QUALIFIED_NAME}/_procedures_/{PROCEDURE_NAME}"
+    )
+    assert sut.schema_qualified_name == SCHEMA_QUALIFIED_NAME
+    assert sut.schema_name == SCHEMA_NAME
+    assert sut.connector_name == CONNECTOR_TYPE
+    assert sut.atlan_schema.qualified_name == SCHEMA_QUALIFIED_NAME