Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

refactor: Use Python context managers for requests #475

Merged
merged 1 commit into from
Mar 19, 2024
Merged

refactor: Use Python context managers for requests #475

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 8 additions & 9 deletions at/utils/authentication.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,14 +24,13 @@ def require_api_key(f, *args, **kwargs):
logger.error('missing api key')
return jsonify(error='API key is missing'), UNAUTHORIZED

response = post(
config['DT_APPAUTH_URL'],
data={'apikey': apikey.strip()})

if response.status_code == OK and response.json()['success'] is True:
logger.debug('valid apikey')
else:
logger.error('invalid api key')
return jsonify(error='API key is invalid'), UNAUTHORIZED
with post(config['DT_APPAUTH_URL'],
data={'apikey': apikey.strip()}) as response:
if (response.status_code == OK and
response.json()['success'] is True):
logger.debug('valid apikey')
else:
logger.error('invalid api key')
return jsonify(error='API key is invalid'), UNAUTHORIZED

return f(*args, **kwargs)
19 changes: 9 additions & 10 deletions at/utils/file.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,22 +102,21 @@ def save_file_from_url(url, upload_dir, logger=getLogger()):
save_filename)

try:
response = get(url)
with get(url) as response:
if response.status_code == OK:
with open(filename, 'w') as file:
file.write(response.text)
else:
logger.error('Error downloading file: {}'.format(url))
raise DownloadError('Error occured while downloading file.')

return (dir_path, filename)
except (ConnectionError, Timeout) as e:
logger.error('Connection error on {url}: {error}'.format(
url=url,
error=e))
raise DownloadError('Error occured while downloading file.')

if response.status_code == OK:
with open(filename, 'w') as file:
file.write(response.text)
else:
logger.error('Error downloading file: {}'.format(url))
raise DownloadError('Error occured while downloading file.')

return (dir_path, filename)


def get_name(filename):
'''Returns draft/rfc name'''
Expand Down
92 changes: 47 additions & 45 deletions at/utils/net.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,71 +41,73 @@ def get_latest(doc, dt_latest_url, logger=getLogger()):
'''Returns URL latest ID/RFC from Datatracker.'''

url = '/'.join([dt_latest_url, doc])
response = get(url)

if response.status_code == OK:
try:
data = response.json()
latest_doc = data['content_url']
except KeyError:
logger.error('can not find content_url for {}'.format(url))
with get(url) as response:
if response.status_code == OK:
try:
data = response.json()
latest_doc = data['content_url']
except KeyError:
logger.error('can not find content_url for {}'.format(url))
raise DocumentNotFound(
'Can not find url for the latest document on '
'datatracker')
else:
logger.error('can not find doc for {}'.format(url))
raise DocumentNotFound(
'Can not find url for the latest document on datatracker')
else:
logger.error('can not find doc for {}'.format(url))
raise DocumentNotFound(
'Can not find the latest document on datatracker')
'Can not find the latest document on datatracker')

return latest_doc
return latest_doc


def get_previous(doc, dt_latest_url, logger=getLogger()):
'''Returns previous ID/RFC from datatracker'''
url = '/'.join([dt_latest_url, doc])
response = get(url)

if response.status_code == OK:
try:
data = response.json()
previous_doc = data['previous']
except KeyError:
logger.error('can not find content_url for {}'.format(url))
with get(url) as response:
if response.status_code == OK:
try:
data = response.json()
previous_doc = data['previous']
except KeyError:
logger.error('can not find content_url for {}'.format(url))
raise DocumentNotFound(
'Can not find url for the previous document on '
'datatracker')
else:
logger.error('can not find doc for {}'.format(url))
raise DocumentNotFound(
'Can not find url for the previous document on datatracker')
else:
logger.error('can not find doc for {}'.format(url))
raise DocumentNotFound(
'Can not find the previous document on datatracker')
'Can not find the previous document on datatracker')

return get_latest(previous_doc, dt_latest_url, logger)
return get_latest(previous_doc, dt_latest_url, logger)


def get_both(doc, dt_latest_url, logger=getLogger()):
'''Returns urls of given doc and previous ID/RFC from Datatracker.'''

url = '/'.join([dt_latest_url, doc])
response = get(url)

if response.status_code == OK:
try:
data = response.json()
latest_doc = data['content_url']
with get(url) as response:
if response.status_code == OK:
try:
previous_doc = data['previous_url']
data = response.json()
latest_doc = data['content_url']
try:
previous_doc = data['previous_url']
except KeyError:
logger.error(
'Can not find previous_url for {}'.format(url))
raise DocumentNotFound(
'Can not find url for previous document on '
'datatracker')
except KeyError:
logger.error('Can not find previous_url for {}'.format(url))
logger.error('can not find content_url for {}'.format(url))
raise DocumentNotFound(
'Can not find url for previous document on datatracker')
except KeyError:
logger.error('can not find content_url for {}'.format(url))
'Can not find url for the latest document on '
'datatracker')
else:
logger.error('can not find doc for {}'.format(url))
raise DocumentNotFound(
'Can not find url for the latest document on datatracker')
else:
logger.error('can not find doc for {}'.format(url))
raise DocumentNotFound(
'Can not find the latest document on datatracker')
'Can not find the latest document on datatracker')

return (previous_doc, latest_doc)
return (previous_doc, latest_doc)


def is_url(string):
Expand Down
Loading