We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[FEATURE]: Handle over 100 secrets when using the `AWSSecretsManager`
### Contact Details [Optional]

support@zenml.io

### Describe the feature you'd like

The [`get_all_secret_keys`](https://github.com/zenml-io/zenml/blob/develop/src/zenml/integrations/aws/secrets_managers/aws_secrets_manager.py#L108) method (found at  `src/zenml/integrations/aws/secrets_managers/aws_secrets_manager.py`) currently has an inbuilt limit whereby it can only access 100 secrets stored in the `AWSSecretsManager`. This has to do with the fact that the API effectively paginates the results. We currently hard-code the max number of results in the method, but ideally:

- we would be able to handle an arbitrary number of secrets (i.e. greater than 100) stored on AWS
- we would no longer hard-code the magic `100` value

### Is your feature request related to a problem?

_No response_

### How do you solve your current problem with the current status-quo of ZenML?

Currently we limit the number of secrets that you can query from AWS to 100.

### Any other comments?

This is related to #ENG-721 and #ENG-720.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/zenml/integrations/aws/secrets_managers/aws_secrets_manager.py`
Content:
```
1 #  Copyright (c) ZenML GmbH 2022. All Rights Reserved.
2 #
3 #  Licensed under the Apache License, Version 2.0 (the "License");
4 #  you may not use this file except in compliance with the License.
5 #  You may obtain a copy of the License at:
6 #
7 #       https://www.apache.org/licenses/LICENSE-2.0
8 #
9 #  Unless required by applicable law or agreed to in writing, software
10 #  distributed under the License is distributed on an "AS IS" BASIS,
11 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
12 #  or implied. See the License for the specific language governing
13 #  permissions and limitations under the License.
14 """Implementation of the AWS Secrets Manager integration."""
15 import json
16 from typing import Any, ClassVar, Dict, List, Optional, cast
17 
18 import boto3
19 
20 from zenml.exceptions import SecretExistsError
21 from zenml.integrations.aws.flavors.aws_secrets_manager_flavor import (
22     AWSSecretsManagerConfig,
23     validate_aws_secret_name_or_namespace,
24 )
25 from zenml.logger import get_logger
26 from zenml.secret.base_secret import BaseSecretSchema
27 from zenml.secrets_managers.base_secrets_manager import (
28     BaseSecretsManager,
29     SecretsManagerScope,
30 )
31 from zenml.secrets_managers.utils import secret_from_dict, secret_to_dict
32 
33 logger = get_logger(__name__)
34 
35 
36 class AWSSecretsManager(BaseSecretsManager):
37     """Class to interact with the AWS secrets manager."""
38 
39     CLIENT: ClassVar[Any] = None
40 
41     @property
42     def config(self) -> AWSSecretsManagerConfig:
43         """Returns the `AWSSecretsManagerConfig` config.
44 
45         Returns:
46             The configuration.
47         """
48         return cast(AWSSecretsManagerConfig, self._config)
49 
50     @classmethod
51     def _ensure_client_connected(cls, region_name: str) -> None:
52         """Ensure that the client is connected to the AWS secrets manager.
53 
54         Args:
55             region_name: the AWS region name
56         """
57         if cls.CLIENT is None:
58             # Create a Secrets Manager client
59             session = boto3.session.Session()
60             cls.CLIENT = session.client(
61                 service_name="secretsmanager", region_name=region_name
62             )
63 
64     def _get_secret_tags(
65         self, secret: BaseSecretSchema
66     ) -> List[Dict[str, str]]:
67         """Return a list of AWS secret tag values for a given secret.
68 
69         Args:
70             secret: the secret object
71 
72         Returns:
73             A list of AWS secret tag values
74         """
75         metadata = self._get_secret_metadata(secret)
76         return [{"Key": k, "Value": v} for k, v in metadata.items()]
77 
78     def _get_secret_scope_filters(
79         self,
80         secret_name: Optional[str] = None,
81     ) -> List[Dict[str, Any]]:
82         """Return a list of AWS filters for the entire scope or just a scoped secret.
83 
84         These filters can be used when querying the AWS Secrets Manager
85         for all secrets or for a single secret available in the configured
86         scope. For more information see: https://docs.aws.amazon.com/secretsmanager/latest/userguide/manage_search-secret.html
87 
88         Example AWS filters for all secrets in the current (namespace) scope:
89 
90         ```python
91         [
92             {
93                 "Key: "tag-key",
94                 "Values": ["zenml_scope"],
95             },
96             {
97                 "Key: "tag-value",
98                 "Values": ["namespace"],
99             },
100             {
101                 "Key: "tag-key",
102                 "Values": ["zenml_namespace"],
103             },
104             {
105                 "Key: "tag-value",
106                 "Values": ["my_namespace"],
107             },
108         ]
109         ```
110 
111         Example AWS filters for a particular secret in the current (namespace)
112         scope:
113 
114         ```python
115         [
116             {
117                 "Key: "tag-key",
118                 "Values": ["zenml_secret_name"],
119             },
120             {
121                 "Key: "tag-value",
122                 "Values": ["my_secret"],
123             },
124             {
125                 "Key: "tag-key",
126                 "Values": ["zenml_scope"],
127             },
128             {
129                 "Key: "tag-value",
130                 "Values": ["namespace"],
131             },
132             {
133                 "Key: "tag-key",
134                 "Values": ["zenml_namespace"],
135             },
136             {
137                 "Key: "tag-value",
138                 "Values": ["my_namespace"],
139             },
140         ]
141         ```
142 
143         Args:
144             secret_name: Optional secret name to filter for.
145 
146         Returns:
147             A list of AWS filters uniquely identifying all secrets
148             or a named secret within the configured scope.
149         """
150         metadata = self._get_secret_scope_metadata(secret_name)
151         filters: List[Dict[str, Any]] = []
152         for k, v in metadata.items():
153             filters.append(
154                 {
155                     "Key": "tag-key",
156                     "Values": [
157                         k,
158                     ],
159                 }
160             )
161             filters.append(
162                 {
163                     "Key": "tag-value",
164                     "Values": [
165                         str(v),
166                     ],
167                 }
168             )
169 
170         return filters
171 
172     def _list_secrets(self, secret_name: Optional[str] = None) -> List[str]:
173         """List all secrets matching a name.
174 
175         This method lists all the secrets in the current scope without loading
176         their contents. An optional secret name can be supplied to filter out
177         all but a single secret identified by name.
178 
179         Args:
180             secret_name: Optional secret name to filter for.
181 
182         Returns:
183             A list of secret names in the current scope and the optional
184             secret name.
185         """
186         self._ensure_client_connected(self.config.region_name)
187 
188         filters: List[Dict[str, Any]] = []
189         prefix: Optional[str] = None
190         if self.config.scope == SecretsManagerScope.NONE:
191             # unscoped (legacy) secrets don't have tags. We want to filter out
192             # non-legacy secrets
193             filters = [
194                 {
195                     "Key": "tag-key",
196                     "Values": [
197                         "!zenml_scope",
198                     ],
199                 },
200             ]
201             if secret_name:
202                 prefix = secret_name
203         else:
204             filters = self._get_secret_scope_filters()
205             if secret_name:
206                 prefix = self._get_scoped_secret_name(secret_name)
207             else:
208                 # add the name prefix to the filters to account for the fact
209                 # that AWS does not do exact matching but prefix-matching on the
210                 # filters
211                 prefix = self._get_scoped_secret_name_prefix()
212 
213         if prefix:
214             filters.append(
215                 {
216                     "Key": "name",
217                     "Values": [
218                         f"{prefix}",
219                     ],
220                 }
221             )
222 
223         # TODO [ENG-720]: Deal with pagination in the aws secret manager when
224         #  listing all secrets
225         # TODO [ENG-721]: take out this magic maxresults number
226         response = self.CLIENT.list_secrets(MaxResults=100, Filters=filters)
227         results = []
228         for secret in response["SecretList"]:
229             name = self._get_unscoped_secret_name(secret["Name"])
230             # keep only the names that are in scope and filter by secret name,
231             # if one was given
232             if name and (not secret_name or secret_name == name):
233                 results.append(name)
234 
235         return results
236 
237     def register_secret(self, secret: BaseSecretSchema) -> None:
238         """Registers a new secret.
239 
240         Args:
241             secret: the secret to register
242 
243         Raises:
244             SecretExistsError: if the secret already exists
245         """
246         validate_aws_secret_name_or_namespace(secret.name)
247         self._ensure_client_connected(self.config.region_name)
248 
249         if self._list_secrets(secret.name):
250             raise SecretExistsError(
251                 f"A Secret with the name {secret.name} already exists"
252             )
253 
254         secret_value = json.dumps(secret_to_dict(secret, encode=False))
255         kwargs: Dict[str, Any] = {
256             "Name": self._get_scoped_secret_name(secret.name),
257             "SecretString": secret_value,
258             "Tags": self._get_secret_tags(secret),
259         }
260 
261         self.CLIENT.create_secret(**kwargs)
262 
263         logger.debug("Created AWS secret: %s", kwargs["Name"])
264 
265     def get_secret(self, secret_name: str) -> BaseSecretSchema:
266         """Gets a secret.
267 
268         Args:
269             secret_name: the name of the secret to get
270 
271         Returns:
272             The secret.
273 
274         Raises:
275             KeyError: if the secret does not exist
276         """
277         validate_aws_secret_name_or_namespace(secret_name)
278         self._ensure_client_connected(self.config.region_name)
279 
280         if not self._list_secrets(secret_name):
281             raise KeyError(f"Can't find the specified secret '{secret_name}'")
282 
283         get_secret_value_response = self.CLIENT.get_secret_value(
284             SecretId=self._get_scoped_secret_name(secret_name)
285         )
286         if "SecretString" not in get_secret_value_response:
287             get_secret_value_response = None
288 
289         return secret_from_dict(
290             json.loads(get_secret_value_response["SecretString"]),
291             secret_name=secret_name,
292             decode=False,
293         )
294 
295     def get_all_secret_keys(self) -> List[str]:
296         """Get all secret keys.
297 
298         Returns:
299             A list of all secret keys
300         """
301         return self._list_secrets()
302 
303     def update_secret(self, secret: BaseSecretSchema) -> None:
304         """Update an existing secret.
305 
306         Args:
307             secret: the secret to update
308 
309         Raises:
310             KeyError: if the secret does not exist
311         """
312         validate_aws_secret_name_or_namespace(secret.name)
313         self._ensure_client_connected(self.config.region_name)
314 
315         if not self._list_secrets(secret.name):
316             raise KeyError(f"Can't find the specified secret '{secret.name}'")
317 
318         secret_value = json.dumps(secret_to_dict(secret))
319 
320         kwargs = {
321             "SecretId": self._get_scoped_secret_name(secret.name),
322             "SecretString": secret_value,
323         }
324 
325         self.CLIENT.put_secret_value(**kwargs)
326 
327     def delete_secret(self, secret_name: str) -> None:
328         """Delete an existing secret.
329 
330         Args:
331             secret_name: the name of the secret to delete
332 
333         Raises:
334             KeyError: if the secret does not exist
335         """
336         self._ensure_client_connected(self.config.region_name)
337 
338         if not self._list_secrets(secret_name):
339             raise KeyError(f"Can't find the specified secret '{secret_name}'")
340 
341         self.CLIENT.delete_secret(
342             SecretId=self._get_scoped_secret_name(secret_name),
343             ForceDeleteWithoutRecovery=True,
344         )
345 
346     def delete_all_secrets(self) -> None:
347         """Delete all existing secrets.
348 
349         This method will force delete all your secrets. You will not be able to
350         recover them once this method is called.
351         """
352         self._ensure_client_connected(self.config.region_name)
353         for secret_name in self._list_secrets():
354             self.CLIENT.delete_secret(
355                 SecretId=self._get_scoped_secret_name(secret_name),
356                 ForceDeleteWithoutRecovery=True,
357             )
358
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/zenml/integrations/aws/secrets_managers/aws_secrets_manager.py b/src/zenml/integrations/aws/secrets_managers/aws_secrets_manager.py
--- a/src/zenml/integrations/aws/secrets_managers/aws_secrets_manager.py
+++ b/src/zenml/integrations/aws/secrets_managers/aws_secrets_manager.py
@@ -32,6 +32,9 @@
 
 logger = get_logger(__name__)
 
+_BOTO_CLIENT_LIST_SECRETS = "list_secrets"
+_PAGINATOR_RESPONSE_SECRETS_LIST_KEY = "SecretList"
+
 
 class AWSSecretsManager(BaseSecretsManager):
     """Class to interact with the AWS secrets manager."""
@@ -220,17 +223,21 @@
                 }
             )
 
-        # TODO [ENG-720]: Deal with pagination in the aws secret manager when
-        #  listing all secrets
-        # TODO [ENG-721]: take out this magic maxresults number
-        response = self.CLIENT.list_secrets(MaxResults=100, Filters=filters)
+        paginator = self.CLIENT.get_paginator(_BOTO_CLIENT_LIST_SECRETS)
+        pages = paginator.paginate(
+            Filters=filters,
+            PaginationConfig={
+                "PageSize": 100,
+            },
+        )
         results = []
-        for secret in response["SecretList"]:
-            name = self._get_unscoped_secret_name(secret["Name"])
-            # keep only the names that are in scope and filter by secret name,
-            # if one was given
-            if name and (not secret_name or secret_name == name):
-                results.append(name)
+        for page in pages:
+            for secret in page[_PAGINATOR_RESPONSE_SECRETS_LIST_KEY]:
+                name = self._get_unscoped_secret_name(secret["Name"])
+                # keep only the names that are in scope and filter by secret name,
+                # if one was given
+                if name and (not secret_name or secret_name == name):
+                    results.append(name)
 
         return results

{"golden_diff": "diff --git a/src/zenml/integrations/aws/secrets_managers/aws_secrets_manager.py b/src/zenml/integrations/aws/secrets_managers/aws_secrets_manager.py\n--- a/src/zenml/integrations/aws/secrets_managers/aws_secrets_manager.py\n+++ b/src/zenml/integrations/aws/secrets_managers/aws_secrets_manager.py\n@@ -32,6 +32,9 @@\n \n logger = get_logger(__name__)\n \n+_BOTO_CLIENT_LIST_SECRETS = \"list_secrets\"\n+_PAGINATOR_RESPONSE_SECRETS_LIST_KEY = \"SecretList\"\n+\n \n class AWSSecretsManager(BaseSecretsManager):\n     \"\"\"Class to interact with the AWS secrets manager.\"\"\"\n@@ -220,17 +223,21 @@\n                 }\n             )\n \n-        # TODO [ENG-720]: Deal with pagination in the aws secret manager when\n-        #  listing all secrets\n-        # TODO [ENG-721]: take out this magic maxresults number\n-        response = self.CLIENT.list_secrets(MaxResults=100, Filters=filters)\n+        paginator = self.CLIENT.get_paginator(_BOTO_CLIENT_LIST_SECRETS)\n+        pages = paginator.paginate(\n+            Filters=filters,\n+            PaginationConfig={\n+                \"PageSize\": 100,\n+            },\n+        )\n         results = []\n-        for secret in response[\"SecretList\"]:\n-            name = self._get_unscoped_secret_name(secret[\"Name\"])\n-            # keep only the names that are in scope and filter by secret name,\n-            # if one was given\n-            if name and (not secret_name or secret_name == name):\n-                results.append(name)\n+        for page in pages:\n+            for secret in page[_PAGINATOR_RESPONSE_SECRETS_LIST_KEY]:\n+                name = self._get_unscoped_secret_name(secret[\"Name\"])\n+                # keep only the names that are in scope and filter by secret name,\n+                # if one was given\n+                if name and (not secret_name or secret_name == name):\n+                    results.append(name)\n \n         return results\n", "issue": "[FEATURE]: Handle over 100 secrets when using the `AWSSecretsManager`\n### Contact Details [Optional]\n\nsupport@zenml.io\n\n### Describe the feature you'd like\n\nThe [`get_all_secret_keys`](https://github.com/zenml-io/zenml/blob/develop/src/zenml/integrations/aws/secrets_managers/aws_secrets_manager.py#L108) method (found at  `src/zenml/integrations/aws/secrets_managers/aws_secrets_manager.py`) currently has an inbuilt limit whereby it can only access 100 secrets stored in the `AWSSecretsManager`. This has to do with the fact that the API effectively paginates the results. We currently hard-code the max number of results in the method, but ideally:\r\n\r\n- we would be able to handle an arbitrary number of secrets (i.e. greater than 100) stored on AWS\r\n- we would no longer hard-code the magic `100` value\n\n### Is your feature request related to a problem?\n\n_No response_\n\n### How do you solve your current problem with the current status-quo of ZenML?\n\nCurrently we limit the number of secrets that you can query from AWS to 100.\n\n### Any other comments?\n\nThis is related to #ENG-721 and #ENG-720.\n", "before_files": [{"content": "#  Copyright (c) ZenML GmbH 2022. All Rights Reserved.\n#\n#  Licensed under the Apache License, Version 2.0 (the \"License\");\n#  you may not use this file except in compliance with the License.\n#  You may obtain a copy of the License at:\n#\n#       https://www.apache.org/licenses/LICENSE-2.0\n#\n#  Unless required by applicable law or agreed to in writing, software\n#  distributed under the License is distributed on an \"AS IS\" BASIS,\n#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express\n#  or implied. See the License for the specific language governing\n#  permissions and limitations under the License.\n\"\"\"Implementation of the AWS Secrets Manager integration.\"\"\"\nimport json\nfrom typing import Any, ClassVar, Dict, List, Optional, cast\n\nimport boto3\n\nfrom zenml.exceptions import SecretExistsError\nfrom zenml.integrations.aws.flavors.aws_secrets_manager_flavor import (\n    AWSSecretsManagerConfig,\n    validate_aws_secret_name_or_namespace,\n)\nfrom zenml.logger import get_logger\nfrom zenml.secret.base_secret import BaseSecretSchema\nfrom zenml.secrets_managers.base_secrets_manager import (\n    BaseSecretsManager,\n    SecretsManagerScope,\n)\nfrom zenml.secrets_managers.utils import secret_from_dict, secret_to_dict\n\nlogger = get_logger(__name__)\n\n\nclass AWSSecretsManager(BaseSecretsManager):\n    \"\"\"Class to interact with the AWS secrets manager.\"\"\"\n\n    CLIENT: ClassVar[Any] = None\n\n    @property\n    def config(self) -> AWSSecretsManagerConfig:\n        \"\"\"Returns the `AWSSecretsManagerConfig` config.\n\n        Returns:\n            The configuration.\n        \"\"\"\n        return cast(AWSSecretsManagerConfig, self._config)\n\n    @classmethod\n    def _ensure_client_connected(cls, region_name: str) -> None:\n        \"\"\"Ensure that the client is connected to the AWS secrets manager.\n\n        Args:\n            region_name: the AWS region name\n        \"\"\"\n        if cls.CLIENT is None:\n            # Create a Secrets Manager client\n            session = boto3.session.Session()\n            cls.CLIENT = session.client(\n                service_name=\"secretsmanager\", region_name=region_name\n            )\n\n    def _get_secret_tags(\n        self, secret: BaseSecretSchema\n    ) -> List[Dict[str, str]]:\n        \"\"\"Return a list of AWS secret tag values for a given secret.\n\n        Args:\n            secret: the secret object\n\n        Returns:\n            A list of AWS secret tag values\n        \"\"\"\n        metadata = self._get_secret_metadata(secret)\n        return [{\"Key\": k, \"Value\": v} for k, v in metadata.items()]\n\n    def _get_secret_scope_filters(\n        self,\n        secret_name: Optional[str] = None,\n    ) -> List[Dict[str, Any]]:\n        \"\"\"Return a list of AWS filters for the entire scope or just a scoped secret.\n\n        These filters can be used when querying the AWS Secrets Manager\n        for all secrets or for a single secret available in the configured\n        scope. For more information see: https://docs.aws.amazon.com/secretsmanager/latest/userguide/manage_search-secret.html\n\n        Example AWS filters for all secrets in the current (namespace) scope:\n\n        ```python\n        [\n            {\n                \"Key: \"tag-key\",\n                \"Values\": [\"zenml_scope\"],\n            },\n            {\n                \"Key: \"tag-value\",\n                \"Values\": [\"namespace\"],\n            },\n            {\n                \"Key: \"tag-key\",\n                \"Values\": [\"zenml_namespace\"],\n            },\n            {\n                \"Key: \"tag-value\",\n                \"Values\": [\"my_namespace\"],\n            },\n        ]\n        ```\n\n        Example AWS filters for a particular secret in the current (namespace)\n        scope:\n\n        ```python\n        [\n            {\n                \"Key: \"tag-key\",\n                \"Values\": [\"zenml_secret_name\"],\n            },\n            {\n                \"Key: \"tag-value\",\n                \"Values\": [\"my_secret\"],\n            },\n            {\n                \"Key: \"tag-key\",\n                \"Values\": [\"zenml_scope\"],\n            },\n            {\n                \"Key: \"tag-value\",\n                \"Values\": [\"namespace\"],\n            },\n            {\n                \"Key: \"tag-key\",\n                \"Values\": [\"zenml_namespace\"],\n            },\n            {\n                \"Key: \"tag-value\",\n                \"Values\": [\"my_namespace\"],\n            },\n        ]\n        ```\n\n        Args:\n            secret_name: Optional secret name to filter for.\n\n        Returns:\n            A list of AWS filters uniquely identifying all secrets\n            or a named secret within the configured scope.\n        \"\"\"\n        metadata = self._get_secret_scope_metadata(secret_name)\n        filters: List[Dict[str, Any]] = []\n        for k, v in metadata.items():\n            filters.append(\n                {\n                    \"Key\": \"tag-key\",\n                    \"Values\": [\n                        k,\n                    ],\n                }\n            )\n            filters.append(\n                {\n                    \"Key\": \"tag-value\",\n                    \"Values\": [\n                        str(v),\n                    ],\n                }\n            )\n\n        return filters\n\n    def _list_secrets(self, secret_name: Optional[str] = None) -> List[str]:\n        \"\"\"List all secrets matching a name.\n\n        This method lists all the secrets in the current scope without loading\n        their contents. An optional secret name can be supplied to filter out\n        all but a single secret identified by name.\n\n        Args:\n            secret_name: Optional secret name to filter for.\n\n        Returns:\n            A list of secret names in the current scope and the optional\n            secret name.\n        \"\"\"\n        self._ensure_client_connected(self.config.region_name)\n\n        filters: List[Dict[str, Any]] = []\n        prefix: Optional[str] = None\n        if self.config.scope == SecretsManagerScope.NONE:\n            # unscoped (legacy) secrets don't have tags. We want to filter out\n            # non-legacy secrets\n            filters = [\n                {\n                    \"Key\": \"tag-key\",\n                    \"Values\": [\n                        \"!zenml_scope\",\n                    ],\n                },\n            ]\n            if secret_name:\n                prefix = secret_name\n        else:\n            filters = self._get_secret_scope_filters()\n            if secret_name:\n                prefix = self._get_scoped_secret_name(secret_name)\n            else:\n                # add the name prefix to the filters to account for the fact\n                # that AWS does not do exact matching but prefix-matching on the\n                # filters\n                prefix = self._get_scoped_secret_name_prefix()\n\n        if prefix:\n            filters.append(\n                {\n                    \"Key\": \"name\",\n                    \"Values\": [\n                        f\"{prefix}\",\n                    ],\n                }\n            )\n\n        # TODO [ENG-720]: Deal with pagination in the aws secret manager when\n        #  listing all secrets\n        # TODO [ENG-721]: take out this magic maxresults number\n        response = self.CLIENT.list_secrets(MaxResults=100, Filters=filters)\n        results = []\n        for secret in response[\"SecretList\"]:\n            name = self._get_unscoped_secret_name(secret[\"Name\"])\n            # keep only the names that are in scope and filter by secret name,\n            # if one was given\n            if name and (not secret_name or secret_name == name):\n                results.append(name)\n\n        return results\n\n    def register_secret(self, secret: BaseSecretSchema) -> None:\n        \"\"\"Registers a new secret.\n\n        Args:\n            secret: the secret to register\n\n        Raises:\n            SecretExistsError: if the secret already exists\n        \"\"\"\n        validate_aws_secret_name_or_namespace(secret.name)\n        self._ensure_client_connected(self.config.region_name)\n\n        if self._list_secrets(secret.name):\n            raise SecretExistsError(\n                f\"A Secret with the name {secret.name} already exists\"\n            )\n\n        secret_value = json.dumps(secret_to_dict(secret, encode=False))\n        kwargs: Dict[str, Any] = {\n            \"Name\": self._get_scoped_secret_name(secret.name),\n            \"SecretString\": secret_value,\n            \"Tags\": self._get_secret_tags(secret),\n        }\n\n        self.CLIENT.create_secret(**kwargs)\n\n        logger.debug(\"Created AWS secret: %s\", kwargs[\"Name\"])\n\n    def get_secret(self, secret_name: str) -> BaseSecretSchema:\n        \"\"\"Gets a secret.\n\n        Args:\n            secret_name: the name of the secret to get\n\n        Returns:\n            The secret.\n\n        Raises:\n            KeyError: if the secret does not exist\n        \"\"\"\n        validate_aws_secret_name_or_namespace(secret_name)\n        self._ensure_client_connected(self.config.region_name)\n\n        if not self._list_secrets(secret_name):\n            raise KeyError(f\"Can't find the specified secret '{secret_name}'\")\n\n        get_secret_value_response = self.CLIENT.get_secret_value(\n            SecretId=self._get_scoped_secret_name(secret_name)\n        )\n        if \"SecretString\" not in get_secret_value_response:\n            get_secret_value_response = None\n\n        return secret_from_dict(\n            json.loads(get_secret_value_response[\"SecretString\"]),\n            secret_name=secret_name,\n            decode=False,\n        )\n\n    def get_all_secret_keys(self) -> List[str]:\n        \"\"\"Get all secret keys.\n\n        Returns:\n            A list of all secret keys\n        \"\"\"\n        return self._list_secrets()\n\n    def update_secret(self, secret: BaseSecretSchema) -> None:\n        \"\"\"Update an existing secret.\n\n        Args:\n            secret: the secret to update\n\n        Raises:\n            KeyError: if the secret does not exist\n        \"\"\"\n        validate_aws_secret_name_or_namespace(secret.name)\n        self._ensure_client_connected(self.config.region_name)\n\n        if not self._list_secrets(secret.name):\n            raise KeyError(f\"Can't find the specified secret '{secret.name}'\")\n\n        secret_value = json.dumps(secret_to_dict(secret))\n\n        kwargs = {\n            \"SecretId\": self._get_scoped_secret_name(secret.name),\n            \"SecretString\": secret_value,\n        }\n\n        self.CLIENT.put_secret_value(**kwargs)\n\n    def delete_secret(self, secret_name: str) -> None:\n        \"\"\"Delete an existing secret.\n\n        Args:\n            secret_name: the name of the secret to delete\n\n        Raises:\n            KeyError: if the secret does not exist\n        \"\"\"\n        self._ensure_client_connected(self.config.region_name)\n\n        if not self._list_secrets(secret_name):\n            raise KeyError(f\"Can't find the specified secret '{secret_name}'\")\n\n        self.CLIENT.delete_secret(\n            SecretId=self._get_scoped_secret_name(secret_name),\n            ForceDeleteWithoutRecovery=True,\n        )\n\n    def delete_all_secrets(self) -> None:\n        \"\"\"Delete all existing secrets.\n\n        This method will force delete all your secrets. You will not be able to\n        recover them once this method is called.\n        \"\"\"\n        self._ensure_client_connected(self.config.region_name)\n        for secret_name in self._list_secrets():\n            self.CLIENT.delete_secret(\n                SecretId=self._get_scoped_secret_name(secret_name),\n                ForceDeleteWithoutRecovery=True,\n            )\n", "path": "src/zenml/integrations/aws/secrets_managers/aws_secrets_manager.py"}], "after_files": [{"content": "#  Copyright (c) ZenML GmbH 2022. All Rights Reserved.\n#\n#  Licensed under the Apache License, Version 2.0 (the \"License\");\n#  you may not use this file except in compliance with the License.\n#  You may obtain a copy of the License at:\n#\n#       https://www.apache.org/licenses/LICENSE-2.0\n#\n#  Unless required by applicable law or agreed to in writing, software\n#  distributed under the License is distributed on an \"AS IS\" BASIS,\n#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express\n#  or implied. See the License for the specific language governing\n#  permissions and limitations under the License.\n\"\"\"Implementation of the AWS Secrets Manager integration.\"\"\"\nimport json\nfrom typing import Any, ClassVar, Dict, List, Optional, cast\n\nimport boto3\n\nfrom zenml.exceptions import SecretExistsError\nfrom zenml.integrations.aws.flavors.aws_secrets_manager_flavor import (\n    AWSSecretsManagerConfig,\n    validate_aws_secret_name_or_namespace,\n)\nfrom zenml.logger import get_logger\nfrom zenml.secret.base_secret import BaseSecretSchema\nfrom zenml.secrets_managers.base_secrets_manager import (\n    BaseSecretsManager,\n    SecretsManagerScope,\n)\nfrom zenml.secrets_managers.utils import secret_from_dict, secret_to_dict\n\nlogger = get_logger(__name__)\n\n_BOTO_CLIENT_LIST_SECRETS = \"list_secrets\"\n_PAGINATOR_RESPONSE_SECRETS_LIST_KEY = \"SecretList\"\n\n\nclass AWSSecretsManager(BaseSecretsManager):\n    \"\"\"Class to interact with the AWS secrets manager.\"\"\"\n\n    CLIENT: ClassVar[Any] = None\n\n    @property\n    def config(self) -> AWSSecretsManagerConfig:\n        \"\"\"Returns the `AWSSecretsManagerConfig` config.\n\n        Returns:\n            The configuration.\n        \"\"\"\n        return cast(AWSSecretsManagerConfig, self._config)\n\n    @classmethod\n    def _ensure_client_connected(cls, region_name: str) -> None:\n        \"\"\"Ensure that the client is connected to the AWS secrets manager.\n\n        Args:\n            region_name: the AWS region name\n        \"\"\"\n        if cls.CLIENT is None:\n            # Create a Secrets Manager client\n            session = boto3.session.Session()\n            cls.CLIENT = session.client(\n                service_name=\"secretsmanager\", region_name=region_name\n            )\n\n    def _get_secret_tags(\n        self, secret: BaseSecretSchema\n    ) -> List[Dict[str, str]]:\n        \"\"\"Return a list of AWS secret tag values for a given secret.\n\n        Args:\n            secret: the secret object\n\n        Returns:\n            A list of AWS secret tag values\n        \"\"\"\n        metadata = self._get_secret_metadata(secret)\n        return [{\"Key\": k, \"Value\": v} for k, v in metadata.items()]\n\n    def _get_secret_scope_filters(\n        self,\n        secret_name: Optional[str] = None,\n    ) -> List[Dict[str, Any]]:\n        \"\"\"Return a list of AWS filters for the entire scope or just a scoped secret.\n\n        These filters can be used when querying the AWS Secrets Manager\n        for all secrets or for a single secret available in the configured\n        scope. For more information see: https://docs.aws.amazon.com/secretsmanager/latest/userguide/manage_search-secret.html\n\n        Example AWS filters for all secrets in the current (namespace) scope:\n\n        ```python\n        [\n            {\n                \"Key: \"tag-key\",\n                \"Values\": [\"zenml_scope\"],\n            },\n            {\n                \"Key: \"tag-value\",\n                \"Values\": [\"namespace\"],\n            },\n            {\n                \"Key: \"tag-key\",\n                \"Values\": [\"zenml_namespace\"],\n            },\n            {\n                \"Key: \"tag-value\",\n                \"Values\": [\"my_namespace\"],\n            },\n        ]\n        ```\n\n        Example AWS filters for a particular secret in the current (namespace)\n        scope:\n\n        ```python\n        [\n            {\n                \"Key: \"tag-key\",\n                \"Values\": [\"zenml_secret_name\"],\n            },\n            {\n                \"Key: \"tag-value\",\n                \"Values\": [\"my_secret\"],\n            },\n            {\n                \"Key: \"tag-key\",\n                \"Values\": [\"zenml_scope\"],\n            },\n            {\n                \"Key: \"tag-value\",\n                \"Values\": [\"namespace\"],\n            },\n            {\n                \"Key: \"tag-key\",\n                \"Values\": [\"zenml_namespace\"],\n            },\n            {\n                \"Key: \"tag-value\",\n                \"Values\": [\"my_namespace\"],\n            },\n        ]\n        ```\n\n        Args:\n            secret_name: Optional secret name to filter for.\n\n        Returns:\n            A list of AWS filters uniquely identifying all secrets\n            or a named secret within the configured scope.\n        \"\"\"\n        metadata = self._get_secret_scope_metadata(secret_name)\n        filters: List[Dict[str, Any]] = []\n        for k, v in metadata.items():\n            filters.append(\n                {\n                    \"Key\": \"tag-key\",\n                    \"Values\": [\n                        k,\n                    ],\n                }\n            )\n            filters.append(\n                {\n                    \"Key\": \"tag-value\",\n                    \"Values\": [\n                        str(v),\n                    ],\n                }\n            )\n\n        return filters\n\n    def _list_secrets(self, secret_name: Optional[str] = None) -> List[str]:\n        \"\"\"List all secrets matching a name.\n\n        This method lists all the secrets in the current scope without loading\n        their contents. An optional secret name can be supplied to filter out\n        all but a single secret identified by name.\n\n        Args:\n            secret_name: Optional secret name to filter for.\n\n        Returns:\n            A list of secret names in the current scope and the optional\n            secret name.\n        \"\"\"\n        self._ensure_client_connected(self.config.region_name)\n\n        filters: List[Dict[str, Any]] = []\n        prefix: Optional[str] = None\n        if self.config.scope == SecretsManagerScope.NONE:\n            # unscoped (legacy) secrets don't have tags. We want to filter out\n            # non-legacy secrets\n            filters = [\n                {\n                    \"Key\": \"tag-key\",\n                    \"Values\": [\n                        \"!zenml_scope\",\n                    ],\n                },\n            ]\n            if secret_name:\n                prefix = secret_name\n        else:\n            filters = self._get_secret_scope_filters()\n            if secret_name:\n                prefix = self._get_scoped_secret_name(secret_name)\n            else:\n                # add the name prefix to the filters to account for the fact\n                # that AWS does not do exact matching but prefix-matching on the\n                # filters\n                prefix = self._get_scoped_secret_name_prefix()\n\n        if prefix:\n            filters.append(\n                {\n                    \"Key\": \"name\",\n                    \"Values\": [\n                        f\"{prefix}\",\n                    ],\n                }\n            )\n\n        paginator = self.CLIENT.get_paginator(_BOTO_CLIENT_LIST_SECRETS)\n        pages = paginator.paginate(\n            Filters=filters,\n            PaginationConfig={\n                \"PageSize\": 100,\n            },\n        )\n        results = []\n        for page in pages:\n            for secret in page[_PAGINATOR_RESPONSE_SECRETS_LIST_KEY]:\n                name = self._get_unscoped_secret_name(secret[\"Name\"])\n                # keep only the names that are in scope and filter by secret name,\n                # if one was given\n                if name and (not secret_name or secret_name == name):\n                    results.append(name)\n\n        return results\n\n    def register_secret(self, secret: BaseSecretSchema) -> None:\n        \"\"\"Registers a new secret.\n\n        Args:\n            secret: the secret to register\n\n        Raises:\n            SecretExistsError: if the secret already exists\n        \"\"\"\n        validate_aws_secret_name_or_namespace(secret.name)\n        self._ensure_client_connected(self.config.region_name)\n\n        if self._list_secrets(secret.name):\n            raise SecretExistsError(\n                f\"A Secret with the name {secret.name} already exists\"\n            )\n\n        secret_value = json.dumps(secret_to_dict(secret, encode=False))\n        kwargs: Dict[str, Any] = {\n            \"Name\": self._get_scoped_secret_name(secret.name),\n            \"SecretString\": secret_value,\n            \"Tags\": self._get_secret_tags(secret),\n        }\n\n        self.CLIENT.create_secret(**kwargs)\n\n        logger.debug(\"Created AWS secret: %s\", kwargs[\"Name\"])\n\n    def get_secret(self, secret_name: str) -> BaseSecretSchema:\n        \"\"\"Gets a secret.\n\n        Args:\n            secret_name: the name of the secret to get\n\n        Returns:\n            The secret.\n\n        Raises:\n            KeyError: if the secret does not exist\n        \"\"\"\n        validate_aws_secret_name_or_namespace(secret_name)\n        self._ensure_client_connected(self.config.region_name)\n\n        if not self._list_secrets(secret_name):\n            raise KeyError(f\"Can't find the specified secret '{secret_name}'\")\n\n        get_secret_value_response = self.CLIENT.get_secret_value(\n            SecretId=self._get_scoped_secret_name(secret_name)\n        )\n        if \"SecretString\" not in get_secret_value_response:\n            get_secret_value_response = None\n\n        return secret_from_dict(\n            json.loads(get_secret_value_response[\"SecretString\"]),\n            secret_name=secret_name,\n            decode=False,\n        )\n\n    def get_all_secret_keys(self) -> List[str]:\n        \"\"\"Get all secret keys.\n\n        Returns:\n            A list of all secret keys\n        \"\"\"\n        return self._list_secrets()\n\n    def update_secret(self, secret: BaseSecretSchema) -> None:\n        \"\"\"Update an existing secret.\n\n        Args:\n            secret: the secret to update\n\n        Raises:\n            KeyError: if the secret does not exist\n        \"\"\"\n        validate_aws_secret_name_or_namespace(secret.name)\n        self._ensure_client_connected(self.config.region_name)\n\n        if not self._list_secrets(secret.name):\n            raise KeyError(f\"Can't find the specified secret '{secret.name}'\")\n\n        secret_value = json.dumps(secret_to_dict(secret))\n\n        kwargs = {\n            \"SecretId\": self._get_scoped_secret_name(secret.name),\n            \"SecretString\": secret_value,\n        }\n\n        self.CLIENT.put_secret_value(**kwargs)\n\n    def delete_secret(self, secret_name: str) -> None:\n        \"\"\"Delete an existing secret.\n\n        Args:\n            secret_name: the name of the secret to delete\n\n        Raises:\n            KeyError: if the secret does not exist\n        \"\"\"\n        self._ensure_client_connected(self.config.region_name)\n\n        if not self._list_secrets(secret_name):\n            raise KeyError(f\"Can't find the specified secret '{secret_name}'\")\n\n        self.CLIENT.delete_secret(\n            SecretId=self._get_scoped_secret_name(secret_name),\n            ForceDeleteWithoutRecovery=True,\n        )\n\n    def delete_all_secrets(self) -> None:\n        \"\"\"Delete all existing secrets.\n\n        This method will force delete all your secrets. You will not be able to\n        recover them once this method is called.\n        \"\"\"\n        self._ensure_client_connected(self.config.region_name)\n        for secret_name in self._list_secrets():\n            self.CLIENT.delete_secret(\n                SecretId=self._get_scoped_secret_name(secret_name),\n                ForceDeleteWithoutRecovery=True,\n            )\n", "path": "src/zenml/integrations/aws/secrets_managers/aws_secrets_manager.py"}]}