Skip to content

Datastores

blackline.models.datastores

DataStore

Bases: BaseSettings

A data model representing a data store, with profiles for connecting to different adapters. # noqa: E501

Parameters:

Name Type Description Default
profiles dict[str, Store] required
name str required
key str required
env_prefix str required
Source code in BAR /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/models/datastores.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
class DataStore(BaseSettings):
    """
    A data model representing a data store, with profiles for connecting to different adapters. # noqa: E501
    """

    profiles: dict[str, Store]
    name: str
    key: str
    env_prefix: str = ""

    class Config:
        extra = "allow"
        env_prefix = ""
        env_nested_delimiter = "__"

    def __getitem__(self, key: str) -> DataStoreBase:
        """
        Retrieves an adapter profile by name.

        Args:
            key (str): The name of the adapter profile to retrieve.

        Raises:
            ValueError: If the specified adapter profile is not found.

        Returns:
            (DataStoreBase): The adapter profile matching the specified key.
        """
        try:
            return self.profiles[key]
        except KeyError:
            raise ValueError(f"Profile {key} not found")

    @root_validator(pre=True)
    def validate_store(cls, values):
        values["key"] = cls.build_key(values)
        return values

    def build_key(values):
        organization_name = values.get("organization_name", "")
        system_name = values.get("system_name", "")
        resource_name = values.get("resource_name", "")
        name = values.get("name", "")
        return f"{organization_name}.{system_name}.{resource_name}.{name}"

__getitem__(key)

Retrieves an adapter profile by name.

Parameters:

Name Type Description Default
key str

The name of the adapter profile to retrieve.

 required

Raises:

Type Description
ValueError

If the specified adapter profile is not found.

Returns:

Type Description
DataStoreBase

The adapter profile matching the specified key.
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/models/datastores.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def __getitem__(self, key: str) -> DataStoreBase:
    """
    Retrieves an adapter profile by name.

    Args:
        key (str): The name of the adapter profile to retrieve.

    Raises:
        ValueError: If the specified adapter profile is not found.

    Returns:
        (DataStoreBase): The adapter profile matching the specified key.
    """
    try:
        return self.profiles[key]
    except KeyError:
        raise ValueError(f"Profile {key} not found")

DataStores

Bases: BaseModel

A data model representing a collection of data stores.

Parameters:

Name Type Description Default
stores list[DataStore] A list of DataStore objects. required
Source code in BAR /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/models/datastores.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
class DataStores(BaseModel):
    """
    A data model representing a collection of data stores.
    """

    stores: list[DataStore] = Field(..., description="A list of DataStore objects.")

    def __getitem__(self, key: str) -> DataStore:
        """
        Retrieves a data store by key.

        Args:
            key (str): The key or name of the data store to retrieve.

        Raises:
            ValueError: If the specified data store is not found.

        Returns:
            (DataStore): The data store matching the specified key.
        """

        for store in self.stores:
            if store.name == key or store.key == key:
                return store
        raise ValueError(f"Store {key} not found")

    def store(self, name: str, profile: Optional[str] = None):
        """
        Retrieves a data store by name and returns its adapter profiles or a specific profile. # noqa: E501

        Args:
            name (str): The name of the data store to retrieve.
            profile (Optional[str], optional): The name of the adapter profile to retrieve. If None, returns all profiles. Defaults to None.  # noqa: E501

        Raises:
            ValueError: If the specified data store is not found.

        Returns:
            (Union[dict[str, Adapter], Adapter]): The adapter profiles for the specified data store, or a specific adapter profile. # noqa: E501
        """
        for store in self.stores:
            if store.name == name:
                if profile:
                    return store.profiles[profile]
                return store.profiles
        raise ValueError(f"Store {name} not found")

    def store_by_key(self, key: str, profile: str) -> DataStoreBase:
        """
        Retrieves a data store by key and returns the specified adapter profile.

        Args:
            key (str): The unique key of the data store to retrieve.
            profile (str): The name of the adapter profile to retrieve.

        Raises:
            ValueError: If the specified data store is not found.

        Returns:
            DataStoreBase: The adapter configuration for the specified data store and profile. # noqa: E501
        """
        for store in self.stores:
            if store.key == key:
                return store.profiles[profile]
        raise ValueError(f"Store {key} not found")

    @classmethod
    def parse_folder(cls, path: Path) -> "DataStores":
        """
        Parses a folder of data store configuration files and adds them to the collection. # noqa: E501

        Args:
            path: The path to the folder containing the data store configuration files.
        """
        _datastores: list[DataStore] = []
        files = config_files(folderpath=path)
        for file in files:
            with open(file, "rb") as f:
                adapter_info = yaml.safe_load(f)
                adapter_info["name"] = file.stem
                adapter_info["organization_name"] = file.parent.parent.parent.name
                adapter_info["system_name"] = file.parent.parent.name
                adapter_info["resource_name"] = file.parent.name
                DataStore.__config__.env_prefix = adapter_info.get("env_prefix", "")
                _DataStore = create_model("_DataStore", __base__=DataStore)
                _DataStore.__name__ = _DataStore.__qualname__ = "DataStore"
                _store = _DataStore.parse_obj(adapter_info)
            _datastores.append(_store)
        return cls(stores=_datastores)

__getitem__(key)

Retrieves a data store by key.

Parameters:

Name Type Description Default
key str

The key or name of the data store to retrieve.

 required

Raises:

Type Description
ValueError

If the specified data store is not found.

Returns:

Type Description
DataStore

The data store matching the specified key.
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/models/datastores.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def __getitem__(self, key: str) -> DataStore:
    """
    Retrieves a data store by key.

    Args:
        key (str): The key or name of the data store to retrieve.

    Raises:
        ValueError: If the specified data store is not found.

    Returns:
        (DataStore): The data store matching the specified key.
    """

    for store in self.stores:
        if store.name == key or store.key == key:
            return store
    raise ValueError(f"Store {key} not found")

parse_folder(path) classmethod

Parses a folder of data store configuration files and adds them to the collection. # noqa: E501

Parameters:

Name Type Description Default
path Path

The path to the folder containing the data store configuration files.

 required
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/models/datastores.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
@classmethod
def parse_folder(cls, path: Path) -> "DataStores":
    """
    Parses a folder of data store configuration files and adds them to the collection. # noqa: E501

    Args:
        path: The path to the folder containing the data store configuration files.
    """
    _datastores: list[DataStore] = []
    files = config_files(folderpath=path)
    for file in files:
        with open(file, "rb") as f:
            adapter_info = yaml.safe_load(f)
            adapter_info["name"] = file.stem
            adapter_info["organization_name"] = file.parent.parent.parent.name
            adapter_info["system_name"] = file.parent.parent.name
            adapter_info["resource_name"] = file.parent.name
            DataStore.__config__.env_prefix = adapter_info.get("env_prefix", "")
            _DataStore = create_model("_DataStore", __base__=DataStore)
            _DataStore.__name__ = _DataStore.__qualname__ = "DataStore"
            _store = _DataStore.parse_obj(adapter_info)
        _datastores.append(_store)
    return cls(stores=_datastores)

store(name, profile=None)

Retrieves a data store by name and returns its adapter profiles or a specific profile. # noqa: E501

Parameters:

Name Type Description Default
name str

The name of the data store to retrieve.

 required
profile Optional[str]

The name of the adapter profile to retrieve. If None, returns all profiles. Defaults to None. # noqa: E501

 None

Raises:

Type Description
ValueError

If the specified data store is not found.

Returns:

Type Description
Union[dict[str, Adapter], Adapter]

The adapter profiles for the specified data store, or a specific adapter profile. # noqa: E501
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/models/datastores.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def store(self, name: str, profile: Optional[str] = None):
    """
    Retrieves a data store by name and returns its adapter profiles or a specific profile. # noqa: E501

    Args:
        name (str): The name of the data store to retrieve.
        profile (Optional[str], optional): The name of the adapter profile to retrieve. If None, returns all profiles. Defaults to None.  # noqa: E501

    Raises:
        ValueError: If the specified data store is not found.

    Returns:
        (Union[dict[str, Adapter], Adapter]): The adapter profiles for the specified data store, or a specific adapter profile. # noqa: E501
    """
    for store in self.stores:
        if store.name == name:
            if profile:
                return store.profiles[profile]
            return store.profiles
    raise ValueError(f"Store {name} not found")

store_by_key(key, profile)

Retrieves a data store by key and returns the specified adapter profile.

Parameters:

Name Type Description Default
key str

The unique key of the data store to retrieve.

 required
profile str

The name of the adapter profile to retrieve.

 required

Raises:

Type Description
ValueError

If the specified data store is not found.

Returns:

Name Type Description
DataStoreBase DataStoreBase

The adapter configuration for the specified data store and profile. # noqa: E501
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/models/datastores.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def store_by_key(self, key: str, profile: str) -> DataStoreBase:
    """
    Retrieves a data store by key and returns the specified adapter profile.

    Args:
        key (str): The unique key of the data store to retrieve.
        profile (str): The name of the adapter profile to retrieve.

    Raises:
        ValueError: If the specified data store is not found.

    Returns:
        DataStoreBase: The adapter configuration for the specified data store and profile. # noqa: E501
    """
    for store in self.stores:
        if store.key == key:
            return store.profiles[profile]
    raise ValueError(f"Store {key} not found")