|
1 | 1 | from __future__ import annotations
|
2 | 2 |
|
3 |
| -from typing import List, Optional |
| 3 | +from typing import Dict, List, Optional |
4 | 4 |
|
5 | 5 | from pyatlan.model.enums import AtlanConnectorType, WorkflowPackage
|
6 | 6 | from pyatlan.model.packages.base.crawler import AbstractCrawler
|
@@ -104,32 +104,34 @@ def basic_auth(
|
104 | 104 | self._credentials_body.update(local_creds)
|
105 | 105 | return self
|
106 | 106 |
|
107 |
| - def include(self, assets: dict) -> MongoDBCrawler: |
| 107 | + def include(self, assets: List[str]) -> MongoDBCrawler: |
108 | 108 | """
|
109 | 109 | Defines the filter for assets to include when crawling.
|
110 | 110 |
|
111 |
| - :param assets: dict where keys are database names, and values are lists of collections. |
| 111 | + :param assets: list of databases names to include when crawling |
112 | 112 | :returns: crawler, set to include only those assets specified
|
113 | 113 | :raises InvalidRequestException: In the unlikely
|
114 | 114 | event the provided filter cannot be translated
|
115 | 115 | """
|
116 |
| - include_assets = assets or {} |
| 116 | + assets = assets or [] |
| 117 | + include_assets: Dict[str, List[str]] = {asset: [] for asset in assets} |
117 | 118 | to_include = self.build_hierarchical_filter(include_assets)
|
118 | 119 | self._parameters.append(
|
119 | 120 | dict(dict(name="include-filter", value=to_include or "{}"))
|
120 | 121 | )
|
121 | 122 | return self
|
122 | 123 |
|
123 |
| - def exclude(self, assets: dict) -> MongoDBCrawler: |
| 124 | + def exclude(self, assets: List[str]) -> MongoDBCrawler: |
124 | 125 | """
|
125 | 126 | Defines the filter for assets to exclude when crawling.
|
126 | 127 |
|
127 |
| - :param assets: dict where keys are database names, and values are lists of collections. |
| 128 | + :param assets: list of databases names to exclude when crawling |
128 | 129 | :returns: crawler, set to exclude only those assets specified
|
129 | 130 | :raises InvalidRequestException: In the unlikely
|
130 | 131 | event the provided filter cannot be translated
|
131 | 132 | """
|
132 |
| - exclude_assets = assets or {} |
| 133 | + assets = assets or [] |
| 134 | + exclude_assets: Dict[str, List[str]] = {asset: [] for asset in assets} |
133 | 135 | to_exclude = self.build_hierarchical_filter(exclude_assets)
|
134 | 136 | self._parameters.append(dict(name="exclude-filter", value=to_exclude or "{}"))
|
135 | 137 | return self
|
|
0 commit comments