sift_py.asset.service

View Source

 1from __future__ import annotations
 2
 3from typing import List, Optional, cast
 4
 5from google.protobuf.field_mask_pb2 import FieldMask
 6from sift.assets.v1.assets_pb2 import (
 7    DeleteAssetRequest,
 8    GetAssetRequest,
 9    GetAssetResponse,
10    UpdateAssetRequest,
11    UpdateAssetResponse,
12)
13from sift.assets.v1.assets_pb2_grpc import AssetServiceStub
14
15from sift_py.asset._internal.shared import list_assets_impl
16from sift_py.asset.config import AssetConfig
17from sift_py.grpc.transport import SiftChannel
18
19
20class AssetService:
21    """
22    A service for managing assets. Allows for creating, updating, and retrieving assets in the Sift API.
23    """
24
25    _asset_service_stub: AssetServiceStub
26
27    def __init__(self, channel: SiftChannel):
28        self._asset_service_stub = AssetServiceStub(channel)
29
30    def get_asset(self, asset_id: str) -> Optional[AssetConfig]:
31        """
32        Retrieves an asset by its ID.
33
34        Args:
35            asset_id: The ID of the asset to retrieve.
36
37        Returns:
38            The Asset if found, None otherwise.
39        """
40        req = GetAssetRequest(asset_id=asset_id)
41        try:
42            res = cast(GetAssetResponse, self._asset_service_stub.GetAsset(req))
43            return AssetConfig.from_asset(res.asset) if res.asset else None
44        except:
45            return None
46
47    def delete_asset(self, asset_id: str) -> None:
48        """
49        Deletes an asset by its ID.
50        """
51        req = DeleteAssetRequest(asset_id=asset_id)
52        self._asset_service_stub.DeleteAsset(req)
53
54    def list_assets(
55        self,
56        names: Optional[List[str]] = None,
57        ids: Optional[List[str]] = None,
58    ) -> List[AssetConfig]:
59        """
60        Lists assets in an organization.
61
62        Args:
63            names: Optional list of names to filter by.
64            ids: Optional list of IDs to filter by.
65
66        Returns:
67            A list of assets matching the criteria.
68        """
69        return [
70            AssetConfig.from_asset(asset)
71            for asset in list_assets_impl(self._asset_service_stub, names, ids)
72        ]
73
74    def update_asset(
75        self, asset: AssetConfig, update_tags: bool = True, update_metadata: bool = True
76    ) -> AssetConfig:
77        """
78        Updates an existing asset.
79
80        Args:
81            asset: The asset to update.
82            update_tags: Whether to update the tags.
83            update_metadata: Whether to update the metadata.
84
85        Returns:
86            The updated AssetConfig.
87        """
88        update_mask = []
89        if update_tags:
90            update_mask.append("tags")
91        if update_metadata:
92            update_mask.append("metadata")
93        req = UpdateAssetRequest(
94            asset=asset.to_asset(),
95            update_mask=FieldMask(paths=update_mask),
96        )
97        res = cast(UpdateAssetResponse, self._asset_service_stub.UpdateAsset(req))
98
99        return AssetConfig.from_asset(res.asset)

class AssetService: View Source

 21class AssetService:
 22    """
 23    A service for managing assets. Allows for creating, updating, and retrieving assets in the Sift API.
 24    """
 25
 26    _asset_service_stub: AssetServiceStub
 27
 28    def __init__(self, channel: SiftChannel):
 29        self._asset_service_stub = AssetServiceStub(channel)
 30
 31    def get_asset(self, asset_id: str) -> Optional[AssetConfig]:
 32        """
 33        Retrieves an asset by its ID.
 34
 35        Args:
 36            asset_id: The ID of the asset to retrieve.
 37
 38        Returns:
 39            The Asset if found, None otherwise.
 40        """
 41        req = GetAssetRequest(asset_id=asset_id)
 42        try:
 43            res = cast(GetAssetResponse, self._asset_service_stub.GetAsset(req))
 44            return AssetConfig.from_asset(res.asset) if res.asset else None
 45        except:
 46            return None
 47
 48    def delete_asset(self, asset_id: str) -> None:
 49        """
 50        Deletes an asset by its ID.
 51        """
 52        req = DeleteAssetRequest(asset_id=asset_id)
 53        self._asset_service_stub.DeleteAsset(req)
 54
 55    def list_assets(
 56        self,
 57        names: Optional[List[str]] = None,
 58        ids: Optional[List[str]] = None,
 59    ) -> List[AssetConfig]:
 60        """
 61        Lists assets in an organization.
 62
 63        Args:
 64            names: Optional list of names to filter by.
 65            ids: Optional list of IDs to filter by.
 66
 67        Returns:
 68            A list of assets matching the criteria.
 69        """
 70        return [
 71            AssetConfig.from_asset(asset)
 72            for asset in list_assets_impl(self._asset_service_stub, names, ids)
 73        ]
 74
 75    def update_asset(
 76        self, asset: AssetConfig, update_tags: bool = True, update_metadata: bool = True
 77    ) -> AssetConfig:
 78        """
 79        Updates an existing asset.
 80
 81        Args:
 82            asset: The asset to update.
 83            update_tags: Whether to update the tags.
 84            update_metadata: Whether to update the metadata.
 85
 86        Returns:
 87            The updated AssetConfig.
 88        """
 89        update_mask = []
 90        if update_tags:
 91            update_mask.append("tags")
 92        if update_metadata:
 93            update_mask.append("metadata")
 94        req = UpdateAssetRequest(
 95            asset=asset.to_asset(),
 96            update_mask=FieldMask(paths=update_mask),
 97        )
 98        res = cast(UpdateAssetResponse, self._asset_service_stub.UpdateAsset(req))
 99
100        return AssetConfig.from_asset(res.asset)

A service for managing assets. Allows for creating, updating, and retrieving assets in the Sift API.

AssetService(channel: grpc.Channel) View Source

28    def __init__(self, channel: SiftChannel):
29        self._asset_service_stub = AssetServiceStub(channel)

def get_asset(self, asset_id: str) -> Union[sift_py.asset.config.AssetConfig, NoneType]: View Source

31    def get_asset(self, asset_id: str) -> Optional[AssetConfig]:
32        """
33        Retrieves an asset by its ID.
34
35        Args:
36            asset_id: The ID of the asset to retrieve.
37
38        Returns:
39            The Asset if found, None otherwise.
40        """
41        req = GetAssetRequest(asset_id=asset_id)
42        try:
43            res = cast(GetAssetResponse, self._asset_service_stub.GetAsset(req))
44            return AssetConfig.from_asset(res.asset) if res.asset else None
45        except:
46            return None

Retrieves an asset by its ID.

Args: asset_id: The ID of the asset to retrieve.

Returns: The Asset if found, None otherwise.

def delete_asset(self, asset_id: str) -> None: View Source

48    def delete_asset(self, asset_id: str) -> None:
49        """
50        Deletes an asset by its ID.
51        """
52        req = DeleteAssetRequest(asset_id=asset_id)
53        self._asset_service_stub.DeleteAsset(req)

Deletes an asset by its ID.

def list_assets( self, names: Union[List[str], NoneType] = None, ids: Union[List[str], NoneType] = None) -> List[sift_py.asset.config.AssetConfig]: View Source

55    def list_assets(
56        self,
57        names: Optional[List[str]] = None,
58        ids: Optional[List[str]] = None,
59    ) -> List[AssetConfig]:
60        """
61        Lists assets in an organization.
62
63        Args:
64            names: Optional list of names to filter by.
65            ids: Optional list of IDs to filter by.
66
67        Returns:
68            A list of assets matching the criteria.
69        """
70        return [
71            AssetConfig.from_asset(asset)
72            for asset in list_assets_impl(self._asset_service_stub, names, ids)
73        ]

Lists assets in an organization.

Args: names: Optional list of names to filter by. ids: Optional list of IDs to filter by.

Returns: A list of assets matching the criteria.

def update_asset( self, asset: sift_py.asset.config.AssetConfig, update_tags: bool = True, update_metadata: bool = True) -> sift_py.asset.config.AssetConfig: View Source

 75    def update_asset(
 76        self, asset: AssetConfig, update_tags: bool = True, update_metadata: bool = True
 77    ) -> AssetConfig:
 78        """
 79        Updates an existing asset.
 80
 81        Args:
 82            asset: The asset to update.
 83            update_tags: Whether to update the tags.
 84            update_metadata: Whether to update the metadata.
 85
 86        Returns:
 87            The updated AssetConfig.
 88        """
 89        update_mask = []
 90        if update_tags:
 91            update_mask.append("tags")
 92        if update_metadata:
 93            update_mask.append("metadata")
 94        req = UpdateAssetRequest(
 95            asset=asset.to_asset(),
 96            update_mask=FieldMask(paths=update_mask),
 97        )
 98        res = cast(UpdateAssetResponse, self._asset_service_stub.UpdateAsset(req))
 99
100        return AssetConfig.from_asset(res.asset)

Updates an existing asset.

Args: asset: The asset to update. update_tags: Whether to update the tags. update_metadata: Whether to update the metadata.

Returns: The updated AssetConfig.