Merge pull request #28 from torbjomg/feat/DP-705-user-asset

feat: user model

pyflake_client/models/assets/user.py

@@ -0,0 +1,39 @@
+"""user"""
+from dataclasses import dataclass
+from typing import Union
+
+from pyflake_client.models.assets.snowflake_asset_interface import ISnowflakeAsset
+from pyflake_client.models.assets.snowflake_principal_interface import ISnowflakePrincipal
+from pyflake_client.models.describables.snowflake_grant_principal import ISnowflakeGrantPrincipal
+from pyflake_client.models.enums.principal import Principal
+
+
+@dataclass(frozen=True)
+class User(ISnowflakeAsset, ISnowflakePrincipal, ISnowflakeGrantPrincipal):
+    """User"""
+
+    name: str
+    comment: str = ""
+    owner: Union[ISnowflakePrincipal, None] = None
+
+    def get_create_statement(self) -> str:
+        """get_create_statement"""
+        if self.owner is None:
+            raise ValueError("Create statement not supported for owner-less users")
+
+        snowflake_principal_type = self.owner.get_snowflake_type().snowflake_type()
+        if snowflake_principal_type not in ["ROLE"]:
+            raise NotImplementedError("Ownership is not implementer for asset of type {self.owner.__class__}")
+
+        return f"""CREATE OR REPLACE USER {self.name} COMMENT = '{self.comment}';
+                   GRANT OWNERSHIP ON USER {self.name} to {snowflake_principal_type} {self.owner.get_identifier()} REVOKE CURRENT GRANTS;"""
+
+    def get_delete_statement(self):
+        """get_delete_statement"""
+        return f"DROP USER IF EXISTS {self.name};"
+
+    def get_identifier(self):
+        return self.name
+
+    def get_snowflake_type(self) -> Principal:
+        return Principal.USER

pyflake_client/models/describables/user.py

@@ -0,0 +1,31 @@
+"""user"""
+from dataclasses import dataclass
+from typing import Union
+import dacite
+
+from pyflake_client.models.describables.snowflake_describable_interface import ISnowflakeDescribable
+from pyflake_client.models.describables.snowflake_grant_principal import ISnowflakeGrantPrincipal
+
+
+@dataclass(frozen=True)
+class User(ISnowflakeDescribable, ISnowflakeGrantPrincipal):
+    """User"""
+
+    name: str
+
+    def get_describe_statement(self) -> str:
+        """get_describe_statement"""
+        return f"SHOW USERS LIKE '{self.name}'".upper()
+
+    def is_procedure(self) -> bool:
+        """is_procedure"""
+        return False
+
+    def get_dacite_config(self) -> Union[dacite.Config, None]:
+        """get_dacite_config"""
+        return None
+
+    @staticmethod
+    def get_snowflake_type() -> str:
+        """get_snowflake_type"""
+        return "USER"

pyflake_client/models/entities/user.py

@@ -0,0 +1,36 @@
+"""user"""
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Union
+
+from pyflake_client.models.entities.snowflake_entity_interface import ISnowflakeEntity
+
+
+@dataclass(frozen=True)
+class User(ISnowflakeEntity):
+    """User"""
+
+    name: str
+    created_on: Union[datetime, None] = None
+    login_name: Union[str, None] = None
+    display_name: Union[str, None] = None
+    first_name: Union[str, None] = None
+    last_name: Union[str, None] = None
+    email: Union[str, None] = None
+    mins_to_unlock: Union[str, None] = None
+    days_to_expiry: Union[str, None] = None
+    comment: Union[str, None] = None
+    disabled: Union[str, None] = None
+    default_warehouse: Union[str, None] = None
+    default_namespace: Union[str, None] = None
+    default_role: Union[str, None] = None
+    deafult_secondary_roles: Union[str, None] = None
+    ext_authn_duo: Union[str, None] = None
+    ext_authn_uid: Union[str, None] = None
+    mins_to_bypass_mfa: Union[str, None] = None
+    owner: Union[str, None] = None
+    last_successful_login: Union[datetime, None] = None
+    expires_at_time: Union[datetime, None] = None
+    locked_until_time: Union[datetime, None] = None
+    has_password: Union[str, None] = None
+    has_rsa_public_key: Union[str, None] = None

pyflake_client/models/enums/principal.py

@@ -1,9 +1,11 @@
 """object_type"""
 from enum import Enum
 
+
 class Principal(str, Enum):
     """Principal"""
-    USER = "TABLE"
+
+    USER = "USER"
     DATABASE_ROLE = "DATABASE_ROLE"
     ROLE = "ROLE"
 
@@ -21,4 +23,4 @@ def grant_type(self) -> str:
         """snowflake_type returns 'ROLE', 'DATABASE_ROLE' or 'USER'"""
         if self == Principal.DATABASE_ROLE:
             return "DATABASE_ROLE"
-        return str(self)
+        return str(self)

pyflake_client/tests/test_user.py

@@ -0,0 +1,58 @@
+"""test_user"""
+import os
+import queue
+import uuid
+from datetime import date
+
+
+from pyflake_client.client import PyflakeClient
+from pyflake_client.models.assets.user import User as UserAsset
+from pyflake_client.models.assets.role import Role as RoleAsset
+from pyflake_client.models.describables.user import User as DescribablesUser
+from pyflake_client.models.entities.user import User as EntitiesUser
+
+
+def test_create_user(flake: PyflakeClient, assets_queue: queue.LifoQueue):
+    """test_create_role"""
+    ### Arrange ###
+    user = UserAsset(
+        name="IGT_CREATE_USER",
+        owner=RoleAsset("USERADMIN"),
+        comment=f"pyflake_client_test_{uuid.uuid4()}",
+    )
+
+    try:
+        flake.register_asset(user, assets_queue)
+
+        ### Act ###
+        sf_user = flake.describe_one(DescribablesUser(user.name), EntitiesUser)
+        ### Assert ###
+        assert sf_user is not None
+        assert sf_user.name == user.name
+        assert sf_user.comment == user.comment
+        assert sf_user.owner == "USERADMIN"
+        assert sf_user.created_on.date() == date.today()
+    finally:
+        ### Cleanup ###
+        flake.delete_assets(assets_queue)
+
+
+def test_get_user(flake: PyflakeClient):
+    """test_get_role"""
+    ### Act ###
+    user_id = os.environ.get("SNOWFLAKE_UID")
+    assert user_id is not None
+    user = flake.describe_one(DescribablesUser(user_id), EntitiesUser)
+
+    ### Assert ###
+    assert user is not None
+    assert user.name == user_id
+
+
+def test_get_user_that_does_not_exist(flake: PyflakeClient):
+    """test_get_role_that_does_not_exist"""
+    ### Act ###
+    user = flake.describe_one(DescribablesUser("I_SURELY_DO_NOT_EXIST"), EntitiesUser)
+
+    ### Assert ###
+    assert user is None