From a9b53b884e74adc83143645ba7b4c5228b14305c Mon Sep 17 00:00:00 2001 From: "Antonio J. Delgado" Date: Mon, 11 Nov 2024 10:46:46 +0200 Subject: [PATCH] Add tries for more requests --- nc_password_client/nc_password_client.py | 133 ++++++++++++++--------- 1 file changed, 82 insertions(+), 51 deletions(-) diff --git a/nc_password_client/nc_password_client.py b/nc_password_client/nc_password_client.py index 42ade66..f7ded3c 100755 --- a/nc_password_client/nc_password_client.py +++ b/nc_password_client/nc_password_client.py @@ -312,21 +312,26 @@ class NextcloudHandler: if src: r = requests.put( f'{self.http}://{self.host}/{path}', - data=open(src, 'rb'), auth=(self.user, self.token), verify=self.ssl, timeout=self.timeout + data=open(src, 'rb'), + auth=(self.user, self.token), + verify=self.ssl, + timeout=self.timeout ) else: r = requests.put( f'{self.http}://{self.host}/{path}', - headers=self.headers, auth=(self.user, self.token), verify=self.ssl, timeout=self.timeout + headers=self.headers, + auth=(self.user, self.token), + verify=self.ssl, + timeout=self.timeout ) if r.status_code in [200, 201, 204]: return r, True - else: - self._log.error( - "Nextcloud instance returned status code: %s", - r.status_code - ) + self._log.error( + "Nextcloud instance returned status code: %s", + r.status_code + ) except requests.exceptions.ReadTimeout as error: self._log.error( "Timeout (%s sec) error doing GET request. %s", @@ -568,8 +573,8 @@ r.status_code else: self._log.error( "Nextcloud instance returned status code: %s", -r.status_code -) + r.status_code + ) def exists_password(self, obj): '''Test if a password exist with the same name''' @@ -583,22 +588,28 @@ r.status_code if 'folder' in post_obj: post_obj['folder'] = self.get_folder_id(post_obj['folder']) if not self.exists_password(post_obj): - r = requests.post( - f'{self.http}://{self.host}/index.php/apps/passwords/api/1.0/password/create', - data=post_obj, - headers=self.headers(), - auth=(self.user, self.token), - verify=self.ssl, timeout=self.timeout - ) + try: + r = requests.post( + f'{self.http}://{self.host}/index.php/apps/passwords/api/1.0/password/create', + data=post_obj, + headers=self.headers(), + auth=(self.user, self.token), + verify=self.ssl, timeout=self.timeout + ) - if r.status_code == 201: - return r.json() - else: + if r.status_code == 201: + return r.json() self._log.error(r.json()) self._log.error( - "Nextcloud instance returned status code: %s", -r.status_code -) + "Nextcloud instance returned status code: %s", + r.status_code + ) + except requests.exceptions.ReadTimeout as error: + self._log.error( + "Timeout (%s sec) error doing GET request. %s", + self.timeout, + error + ) else: self._log.warning( "Password with name '%s' already exists", @@ -607,39 +618,54 @@ r.status_code def delete_password(self, post_obj): '''Delete a password''' - r = requests.delete( - f'{self.http}://{self.host}/index.php/apps/passwords/api/1.0/password/delete', - data=post_obj, - headers=self.headers(), - auth=(self.user, self.token), - verify=self.ssl, timeout=self.timeout - ) + try: + r = requests.delete( + f'{self.http}://{self.host}/index.php/apps/passwords/api/1.0/password/delete', + data=post_obj, + headers=self.headers(), + auth=(self.user, self.token), + verify=self.ssl, timeout=self.timeout + ) - if r.status_code == 200: - return r.json() - else: + if r.status_code == 200: + return r.json() + else: + self._log.error( + "Nextcloud instance returned status code: %s", + r.status_code + ) + except requests.exceptions.ReadTimeout as error: self._log.error( - "Nextcloud instance returned status code: %s", -r.status_code -) + "Timeout (%s sec) error doing GET request. %s", + self.timeout, + error + ) + def update_password(self, post_obj): '''Update a password''' - r = requests.patch( - f'{self.http}://{self.host}/index.php/apps/passwords/api/1.0/password/update', - data=post_obj, - headers=self.headers(), - auth=(self.user, self.token), - verify=self.ssl, timeout=self.timeout - ) + try: + r = requests.patch( + f'{self.http}://{self.host}/index.php/apps/passwords/api/1.0/password/update', + data=post_obj, + headers=self.headers(), + auth=(self.user, self.token), + verify=self.ssl, timeout=self.timeout + ) - if r.status_code == 200: - return r.json() - else: + if r.status_code == 200: + return r.json() + else: + self._log.error( + "Nextcloud instance returned status code: %s", + r.status_code + ) + except requests.exceptions.ReadTimeout as error: self._log.error( - "Nextcloud instance returned status code: %s", -r.status_code -) + "Timeout (%s sec) error doing GET request. %s", + self.timeout, + error + ) def is_same_key(self, key, dict1, dict2): '''Test if two dictionaries have the same key with the same value''' @@ -653,9 +679,10 @@ r.status_code if obj1 == obj2: return True if ( - self.is_same_key('username', obj1, obj2) and + self.is_same_key('username', obj1, obj2) and self.is_same_key('password', obj1, obj2) and - self.is_same_key('url', obj1, obj2) + self.is_same_key('url', obj1, obj2) and + self.is_same_key('folder', obj1, obj2) ): return True return False @@ -910,7 +937,11 @@ def delete_passwords_folder(ctx, name): ctx.obj['NcPasswordClient'].delete_passwords_folder(name) @cli.command() -@click.option('--limit', '-l', default=-1, help='Maximun number of passwords to migrate. -1 for unlimited.') +@click.option( + '--limit', '-l', + default=-1, + help='Maximun number of passwords to migrate. -1 for unlimited.' +) @click_config_file.configuration_option() @click.pass_context def migrate_pass(ctx, limit):