cirro.sdk.asset

 1import fnmatch
 2from abc import abstractmethod
 3from typing import List, TypeVar
 4
 5from cirro.sdk.exceptions import DataPortalAssetNotFound, DataPortalInputError
 6
 7
 8class DataPortalAsset:
 9    """Base class used for all Data Portal Assets"""
10
11    @property
12    @abstractmethod
13    def name(self):
14        """Asset name"""
15        pass
16
17    def __repr__(self):
18        return f'{self.__class__.__name__}(name={self.name})'
19
20
21T = TypeVar('T', bound=DataPortalAsset)
22
23
24class DataPortalAssets(List[T]):
25    """
26    Generic class with helper functions for any group of assets (projects, datasets, etc.)
27    """
28
29    # Overridden by child classes
30    asset_name = 'asset'
31
32    def __init__(self, input_list: List[T]):
33        super().__init__(input_list)
34
35    def __str__(self):
36        return "\n\n".join([str(i) for i in self])
37
38    def description(self):
39        """Render a text summary of the assets."""
40
41        return '\n\n---\n\n'.join([
42            str(i)
43            for i in self
44        ])
45
46    def get_by_name(self, name: str) -> T:
47        """Return the item which matches with name attribute."""
48
49        if name is None:
50            raise DataPortalInputError(f"Must provide name to identify {self.asset_name}")
51
52        # Get the items which have a matching name
53        matching_queries = [i for i in self if i.name == name]
54
55        # Error if no items are found
56        msg = '\n'.join([f"No {self.asset_name} found with name '{name}'.", self.description()])
57        if len(matching_queries) == 0:
58            raise DataPortalAssetNotFound(msg)
59
60        # Error if multiple projects are found
61        msg = f"Multiple {self.asset_name} items found with name '{name}', use ID instead.\n{self.description()}"
62        if len(matching_queries) > 1:
63            raise DataPortalAssetNotFound(msg)
64
65        return matching_queries[0]
66
67    def get_by_id(self, _id: str) -> T:
68        """Return the item which matches by id attribute."""
69
70        if _id is None:
71            raise DataPortalInputError(f"Must provide id to identify {self.asset_name}")
72
73        # Get the items which have a matching ID
74        matching_queries = [i for i in self if i.id == _id]
75
76        # Error if no items are found
77        msg = '\n'.join([f"No {self.asset_name} found with id '{_id}'.", self.description()])
78        if len(matching_queries) == 0:
79            raise DataPortalAssetNotFound(msg)
80
81        return matching_queries[0]
82
83    def filter_by_pattern(self, pattern: str) -> 'DataPortalAssets[T]':
84        """Filter the items to just those whose name attribute matches the pattern."""
85
86        # Get a list of the names to search against
87        all_names = [i.name for i in self]
88
89        # Filter the names by the pattern
90        filtered_names = fnmatch.filter(all_names, pattern)
91
92        # Filter the list to only include those items
93        return self.__class__(
94            [
95                i
96                for i in self
97                if i.name in filtered_names
98            ]
99        )