[DOP-24139] Add method for creating namespaces

conftest.py

@@ -15,7 +15,6 @@
     KeyValueIntHWM,
 )
 from horizon.client.auth import LoginPassword
-from horizon.commons.schemas.v1 import NamespaceCreateRequestV1
 from packaging.version import Version
 
 from horizon_hwm_store import HorizonHWMStore
@@ -149,6 +148,11 @@ def file_with_mtime(mtime: datetime) -> Path:
 
 
 @pytest.fixture(params=HWMS_WITH_VALUE)
+def hwm_new_values(request):
+    return request.param
+
+
+@pytest.fixture(params=[HWMS_WITH_VALUE[0]])
 def hwm_new_value(request):
     return request.param
 
@@ -164,16 +168,9 @@ def hwm_store():
 
 @pytest.fixture(scope="module")
 def ensure_namespace():
-    from requests.exceptions import HTTPError
 
-    store = HorizonHWMStore(
+    HorizonHWMStore(
         api_url=HORIZON_URL,
         auth=LoginPassword(login=HORIZON_USER, password=HORIZON_PASSWORD),
         namespace=HORIZON_NAMESPACE,
-    )
-
-    try:
-        store.client.create_namespace(NamespaceCreateRequestV1(name=HORIZON_NAMESPACE))  # noqa: WPS437
-    except HTTPError:
-        # exception: 409 Client Error: Conflict for url: http://horizon/v1/namespaces/ - namespace already exists
-        pass
+    ).force_create_namespace()

docs/changelog/next_release/44.feature.rst

@@ -0,0 +1 @@
+Add ``force_create_namespace`` method

docs/horizon-hwm-store.rst

@@ -6,4 +6,4 @@ Horizon HWM Store
 .. currentmodule:: horizon_hwm_store.horizon_hwm_store
 
 .. autoclass:: HorizonHWMStore
-    :members: get_hwm, set_hwm
+    :members: get_hwm, set_hwm, force_create_namespace, check

horizon_hwm_store/horizon_hwm_store.py

@@ -8,11 +8,14 @@
 from etl_entities.hwm_store import BaseHWMStore, register_hwm_store_class
 from horizon.client.auth import LoginPassword
 from horizon.client.sync import HorizonClientSync, RetryConfig, TimeoutConfig
+from horizon.commons.exceptions import EntityAlreadyExistsError
 from horizon.commons.schemas.v1 import (
     HWMCreateRequestV1,
     HWMPaginateQueryV1,
     HWMUpdateRequestV1,
+    NamespaceCreateRequestV1,
     NamespacePaginateQueryV1,
+    NamespaceResponseV1,
 )
 
 try:
@@ -33,7 +36,7 @@
 
     .. warning::
 
-        It is required to create namespace in Horizon BEFORE using this class.
+        It is required to create a namespace BEFORE working with HWMs.
 
     Parameters
     ----------
@@ -98,7 +101,7 @@
             namespace="namespace",
             retry=RetryConfig(total=5),
             timeout=TimeoutConfig(request_timeout=10),
-        ):
+        ).force_create_namespace():
             with IncrementalStrategy():
                 df = reader.run()
                 writer.run(df)
@@ -213,6 +216,24 @@
         self._get_namespace_id()
         return self
 
+    def force_create_namespace(self) -> HorizonHWMStore:
+        """
+        Create a namespace with name specified in HorizonHWMStore class.
+
+        Returns
+        -------
+        HorizonHWMStore
+            Self
+        """
+        namespace = self._get_namespace(self.namespace)
+        if namespace is None:
+            try:
+                namespace = self.client.create_namespace(NamespaceCreateRequestV1(name=self.namespace))
+                self._namespace_id = namespace.id  # noqa: WPS601
+            except EntityAlreadyExistsError:
+                ...
+        return self
+
     # LoginPassword, RetryConfig and TimeoutConfig can be inherited from Pydantic v2 BaseModel
     # which is detected by Pydantic v1 as arbitrary type. So we need to parse them manually.
     @validator("auth", pre=True)
@@ -233,6 +254,10 @@
             return TimeoutConfig.parse_obj(value)
         return value
 
+    def _get_namespace(self, name: str) -> NamespaceResponseV1 | None:
+        namespaces = self.client.paginate_namespaces(query=NamespacePaginateQueryV1(name=name)).items
+        return namespaces[0] if namespaces else None
+
     def _get_namespace_id(self) -> int:
         """
         Fetch the ID of the namespace. Raises an exception if the namespace doesn't exist.
@@ -250,11 +275,14 @@
         if self._namespace_id is not None:
             return self._namespace_id
 
-        namespaces = self.client.paginate_namespaces(NamespacePaginateQueryV1(name=self.namespace)).items
-        if not namespaces:
-            raise RuntimeError(f"Namespace {self.namespace!r} not found. Please create it before using.")
+        namespace = self._get_namespace(self.namespace)
+        if namespace is None:
+            raise RuntimeError(
+                f"Namespace {self.namespace!r} not found. "
+                "Please create it before using by calling .force_create_namespace() method.",
+            )
 
-        self._namespace_id = namespaces[0].id  # noqa: WPS601
+        self._namespace_id = namespace.id  # noqa: WPS601
         return self._namespace_id
 
     def _get_hwm_id(self, namespace_id: int, hwm_name: str) -> Optional[int]:

tests/tests_integration/test_horizon_hwm_store_integration.py

@@ -14,8 +14,8 @@
 HORIZON_NAMESPACE = os.environ.get("HORIZON_NAMESPACE")
 
 
-def test_hwm_store_integration(hwm_store, hwm_new_value, ensure_namespace):
-    hwm, new_value = hwm_new_value
+def test_hwm_store_integration(hwm_store, hwm_new_values, ensure_namespace):
+    hwm, new_value = hwm_new_values
     assert hwm_store.get_hwm(hwm.name) is None
 
     hwm_store.set_hwm(hwm)
@@ -65,3 +65,19 @@ def test_hwm_store_unexisting_namespace(hwm_new_value):
 
     with pytest.raises(RuntimeError, match=error_msg):
         store.set_hwm(hwm)
+
+
+def test_hwm_store_creates_namespace():
+    namespace_name = "new_namespace"
+    store = HorizonHWMStore(
+        api_url=HORIZON_URL,
+        auth=LoginPassword(login=HORIZON_USER, password=HORIZON_PASSWORD),
+        namespace=namespace_name,
+    )
+    error_msg = "Namespace 'new_namespace' not found. Please create it before using."
+
+    with pytest.raises(RuntimeError, match=error_msg):
+        store.get_hwm("some_hwm_name")
+
+    store.force_create_namespace()
+    assert store.get_hwm("some_hwm_name") is None