deepblue/GoogleDriveManagement
1
0
Fork 0
mirror of https://github.com/GAM-team/GAM.git synced 2025-07-09 06:03:35 +00:00
Code Issues Actions 19 Packages Projects Releases Wiki Activity

pyupgrade --py37-plus **/*.py (#1445)

Browse Source
This commit is contained in:
Janosh Riebesell
2022-01-11 17:05:02 +01:00
committed by GitHub
parent 676908daca
commit 19f21a9453
20 changed files with 50 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/gam.py
View File

@ -1,5 +1,4 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Provides backwards compatibility for calling gam as a single .py file"""
import sys

10
src/gam/__init__.py
View File

@ -774,12 +774,12 @@ def doGAMVersion(checkForArgs=True):
cpu_bits = struct.calcsize('P') * 8
api_client_ver = lib_version('google-api-python-client')
print(
(f'GAM {GAM_VERSION} - {GAM_URL} - {GM_Globals[GM_GAM_TYPE]}\n'
f'GAM {GAM_VERSION} - {GAM_URL} - {GM_Globals[GM_GAM_TYPE]}\n'
f'{GAM_AUTHOR}\n'
f'Python {pyversion} {cpu_bits}-bit {sys.version_info.releaselevel}\n'
f'google-api-python-client {api_client_ver}\n'
f'{getOSPlatform()} {platform.machine()}\n'
f'Path: {GM_Globals[GM_GAM_PATH]}'))
f'Path: {GM_Globals[GM_GAM_PATH]}')
if sys.platform.startswith('win') and \
cpu_bits == 32 and \
platform.machine().find('64') != -1:
@ -4098,7 +4098,7 @@ def downloadDriveFile(users):
for sheet in spreadsheet['sheets']:
if sheet['properties']['title'].lower(
) == csvSheetTitleLower:
spreadsheetUrl = '{0}?format=csv&id={1}&gid={2}'.format(
spreadsheetUrl = '{}?format=csv&id={}&gid={}'.format(
re.sub('/edit.*$', '/export',
spreadsheet['spreadsheetUrl']),
fileId, sheet['properties']['sheetId'])
@ -4123,7 +4123,7 @@ def downloadDriveFile(users):
while not done:
status, done = downloader.next_chunk()
if showProgress:
print('Downloaded: {0:>7.2%}'.format(
print('Downloaded: {:>7.2%}'.format(
status.progress()))
else:
_, content = drive._http.request(uri=spreadsheetUrl,
@ -4135,7 +4135,7 @@ def downloadDriveFile(users):
fileutils.close_file(fh)
fileDownloaded = True
break
except (IOError, httplib2.HttpLib2Error) as e:
except (OSError, httplib2.HttpLib2Error) as e:
display.print_error(str(e))
GM_Globals[GM_SYSEXITRC] = 6
fileDownloadFailed = True

1
src/gam/__main__.py
View File

@ -1,5 +1,4 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# GAM
#

2
src/gam/auth/__init__.py
View File

@ -37,7 +37,7 @@ def get_admin_credentials(api=None):
credential_file = get_admin_credentials_filename()
if not os.path.isfile(credential_file):
raise oauth.InvalidCredentialsFileError
with open(credential_file, 'r') as f:
with open(credential_file) as f:
creds_data = json.load(f)
# Validate that enable DASA matches content of authorization file
if GC_Values[GC_ENABLE_DASA] and 'private_key_id' in creds_data:

12
src/gam/auth/oauth.py
View File

@ -115,7 +115,7 @@ class Credentials(google.oauth2.credentials.Credentials):
Raises:
TypeError: If id_token_data is not the required dict type.
"""
super(Credentials, self).__init__(token=token,
super().__init__(token=token,
refresh_token=refresh_token,
id_token=id_token,
token_uri=token_uri,
@ -161,9 +161,9 @@ class Credentials(google.oauth2.credentials.Credentials):
ValueError: If missing fields are detected in the info.
"""
# We need all of these keys
keys_needed = set(('client_id', 'client_secret'))
keys_needed = {'client_id', 'client_secret'}
# We need 1 or more of these keys
keys_need_one_of = set(('refresh_token', 'auth_token', 'token'))
keys_need_one_of = {'refresh_token', 'auth_token', 'token'}
missing = keys_needed.difference(info.keys())
has_one_of = set(info) & keys_need_one_of
if missing or not has_one_of:
@ -472,7 +472,7 @@ class Credentials(google.oauth2.credentials.Credentials):
def _locked_refresh(self, request):
"""Refreshes the credential's access token while the file lock is held."""
assert self._lock.is_locked
super(Credentials, self).refresh(request)
super().refresh(request)
def write(self):
"""Writes credentials to disk."""
@ -523,12 +523,12 @@ class _ShortURLFlow(google_auth_oauthlib.flow.InstalledAppFlow):
def authorization_url(self, http=None, **kwargs):
"""Gets a shortened authorization URL."""
long_url, state = super(_ShortURLFlow, self).authorization_url(**kwargs)
long_url, state = super().authorization_url(**kwargs)
short_url = utils.shorten_url(long_url)
return short_url, state
class _FileLikeThreadLock(object):
class _FileLikeThreadLock:
"""A threading.lock which has the same interface as filelock.Filelock."""
def __init__(self):

6
src/gam/auth/oauth_test.py
View File

@ -38,7 +38,7 @@ class CredentialsTest(unittest.TestCase):
'client_id': self.fake_client_id,
'client_secret': self.fake_client_secret,
}
super(CredentialsTest, self).setUp()
super().setUp()
def tearDown(self):
# Remove any credential files that may have been created.
@ -46,7 +46,7 @@ class CredentialsTest(unittest.TestCase):
os.remove(self.fake_filename)
if os.path.exists('%s.lock' % self.fake_filename):
os.remove('%s.lock' % self.fake_filename)
super(CredentialsTest, self).tearDown()
super().tearDown()
def test_from_authorized_user_info_only_required_info(self):
creds = oauth.Credentials.from_authorized_user_info(
@ -592,7 +592,7 @@ class ShortUrlFlowTest(unittest.TestCase):
}
self.long_url = 'http://example.com/some/long/url'
self.short_url = 'http://ex.co/short'
super(ShortUrlFlowTest, self).setUp()
super().setUp()
@patch.object(oauth.google_auth_oauthlib.flow.InstalledAppFlow,
'authorization_url')

4
src/gam/controlflow.py
View File

@ -92,8 +92,8 @@ def wait_on_failure(current_attempt_num,
wait_on_fail = min(2**current_attempt_num,
60) + float(random.randint(1, 1000)) / 1000
if current_attempt_num > error_print_threshold:
sys.stderr.write((f'Temporary error: {error_message}, Backing off: '
sys.stderr.write(f'Temporary error: {error_message}, Backing off: '
f'{int(wait_on_fail)} seconds, Retry: '
f'{current_attempt_num}/{total_num_retries}\n'))
f'{current_attempt_num}/{total_num_retries}\n')
sys.stderr.flush()
time.sleep(wait_on_fail)

8
src/gam/display.py
View File

@ -259,9 +259,9 @@ def write_csv_file(csvRows, titles, list_type, todrive):
dialect='nixstdout',
extrasaction='ignore')
try:
writer.writerow(dict((item, item) for item in writer.fieldnames))
writer.writerow({item: item for item in writer.fieldnames})
writer.writerows(csvRows)
except IOError as e:
except OSError as e:
controlflow.system_error_exit(6, e)
if todrive:
admin_email = gam._get_admin_email()
@ -309,12 +309,12 @@ and follow recommend steps to authorize GAM for Drive access.''')
def print_error(message):
"""Prints a one-line error message to stderr in a standard format."""
sys.stderr.write('\n{0}{1}\n'.format(ERROR_PREFIX, message))
sys.stderr.write(f'\n{ERROR_PREFIX}{message}\n')
def print_warning(message):
"""Prints a one-line warning message to stderr in a standard format."""
sys.stderr.write('\n{0}{1}\n'.format(WARNING_PREFIX, message))
sys.stderr.write(f'\n{WARNING_PREFIX}{message}\n')
def print_json(object_value, spacing=''):

10
src/gam/fileutils.py
View File

@ -59,7 +59,7 @@ def open_file(filename,
# Open a file on disk
f = _open_file(filename, mode, newline=newline, encoding=encoding)
if strip_utf_bom:
utf_bom = u'\ufeff'
utf_bom = '\ufeff'
has_bom = False
if 'b' in mode:
@ -79,7 +79,7 @@ def open_file(filename,
return f
except IOError as e:
except OSError as e:
controlflow.system_error_exit(6, e)
@ -101,7 +101,7 @@ def close_file(f, force_flush=False):
try:
f.close()
return True
except IOError as e:
except OSError as e:
display.print_error(e)
return False
@ -140,7 +140,7 @@ def read_file(filename,
encoding=encoding) as f:
return f.read()
except IOError as e:
except OSError as e:
if continue_on_error:
if display_errors:
display.print_warning(e)
@ -174,7 +174,7 @@ def write_file(filename,
f.write(data)
return True
except IOError as e:
except OSError as e:
if continue_on_error:
if display_errors:
display.print_error(e)

6
src/gam/fileutils_test.py
View File

@ -13,7 +13,7 @@ class FileutilsTest(unittest.TestCase):
def setUp(self):
self.fake_path = '/some/path/to/file'
super(FileutilsTest, self).setUp()
super().setUp()
@patch.object(fileutils.sys, 'stdin')
def test_open_file_stdin(self, mock_stdin):
@ -63,7 +63,7 @@ class FileutilsTest(unittest.TestCase):
self.assertEqual(fileutils.UTF8_SIG, mock_open.call_args[1]['encoding'])
def test_open_file_strips_utf_bom_in_utf(self):
bom_prefixed_data = u'\ufefffoobar'
bom_prefixed_data = '\ufefffoobar'
fake_file = io.StringIO(bom_prefixed_data)
mock_open = MagicMock(spec=open, return_value=fake_file)
with patch.object(fileutils, 'open', mock_open):
@ -89,7 +89,7 @@ class FileutilsTest(unittest.TestCase):
self.assertEqual('foobar', f.read())
def test_open_file_strips_utf_bom_in_binary(self):
bom_prefixed_data = u'\ufefffoobar'.encode('UTF-8')
bom_prefixed_data = '\ufefffoobar'.encode()
fake_file = io.BytesIO(bom_prefixed_data)
mock_open = MagicMock(spec=open, return_value=fake_file)
with patch.object(fileutils, 'open', mock_open):

2
src/gam/gapi/__init___test.py
View File

@ -80,7 +80,7 @@ class GapiTest(unittest.TestCase):
]
self.empty_items_response = {'items': []}
super(GapiTest, self).setUp()
super().setUp()
def test_call_returns_basic_200_response(self):
response = gapi.call(self.mock_service, self.mock_method_name)

4
src/gam/gapi/directory/cros.py
View File

@ -755,11 +755,11 @@ def doPrintCrosDevices():
cros['autoUpdateExpiration'])
row = {}
for attrib in cros:
if attrib not in set([
if attrib not in {
'kind', 'etag', 'tpmVersionInfo', 'recentUsers',
'activeTimeRanges', 'deviceFiles', 'cpuStatusReports',
'diskVolumeReports', 'systemRamFreeReports'
]):
}:
row[attrib] = cros[attrib]
if selectedLists.get('activeTimeRanges'):
timergs = cros.get('activeTimeRanges', [])

2
src/gam/gapi/directory/orgunits.py
View File

@ -299,7 +299,7 @@ def update():
def orgUnitPathQuery(path, checkSuspended):
query = "orgUnitPath='{0}'".format(path.replace(
query = "orgUnitPath='{}'".format(path.replace(
"'", "\\'")) if path != '/' else ''
if checkSuspended is not None:
query += f' isSuspended={checkSuspended}'

8
src/gam/gapi/reports.py
View File

@ -504,9 +504,9 @@ def showReport():
purge_parameters = True
for event in events:
for item in event.get('parameters', []):
if set(item) == set(['value', 'name']):
if set(item) == {'value', 'name'}:
event[item['name']] = item['value']
elif set(item) == set(['intValue', 'name']):
elif set(item) == {'intValue', 'name'}:
if item['name'] in ['start_time', 'end_time']:
val = item.get('intValue')
if val is not None:
@ -517,9 +517,9 @@ def showReport():
val-62135683200).isoformat()
else:
event[item['name']] = item['intValue']
elif set(item) == set(['boolValue', 'name']):
elif set(item) == {'boolValue', 'name'}:
event[item['name']] = item['boolValue']
elif set(item) == set(['multiValue', 'name']):
elif set(item) == {'multiValue', 'name'}:
event[item['name']] = ' '.join(item['multiValue'])
elif item['name'] == 'scope_data':
parts = {}

2
src/gam/gapi/vault.py
View File

@ -790,7 +790,7 @@ def downloadExport():
done = False
while not done:
status, done = downloader.next_chunk()
sys.stdout.write(' Downloaded: {0:>7.2%}\r'.format(
sys.stdout.write(' Downloaded: {:>7.2%}\r'.format(
status.progress()))
sys.stdout.flush()
sys.stdout.write('\n Download complete. Flushing to disk...\n')

4
src/gam/transport.py
View File

@ -90,7 +90,7 @@ class Request(google_auth_httplib2.Request):
@_force_user_agent(GAM_USER_AGENT)
def __call__(self, *args, **kwargs):
"""Inserts the GAM user-agent header in requests."""
return super(Request, self).__call__(*args, **kwargs)
return super().__call__(*args, **kwargs)
class AuthorizedHttp(google_auth_httplib2.AuthorizedHttp):
@ -99,4 +99,4 @@ class AuthorizedHttp(google_auth_httplib2.AuthorizedHttp):
@_force_user_agent(GAM_USER_AGENT)
def request(self, *args, **kwargs):
"""Inserts the GAM user-agent header in requests."""
return super(AuthorizedHttp, self).request(*args, **kwargs)
return super().request(*args, **kwargs)

4
src/gam/transport_test.py
View File

@ -15,7 +15,7 @@ class CreateHttpTest(unittest.TestCase):
def setUp(self):
SetGlobalVariables()
super(CreateHttpTest, self).setUp()
super().setUp()
def test_create_http_sets_default_values_on_http(self):
http = transport.create_http()
@ -56,7 +56,7 @@ class TransportTest(unittest.TestCase):
self.mock_content)
self.mock_credentials = MagicMock()
self.test_uri = 'http://example.com'
super(TransportTest, self).setUp()
super().setUp()
@patch.object(transport, 'create_http')
def test_create_request_uses_default_http(self, mock_create_http):

8
src/gam/utils.py
View File

@ -1,7 +1,3 @@
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import datetime
import re
import sys
@ -32,7 +28,7 @@ class LazyLoader(types.ModuleType):
self._local_name = local_name
self._parent_module_globals = parent_module_globals
super(LazyLoader, self).__init__(name)
super().__init__(name)
def _load(self):
# Import the target module and insert it into the parent's namespace
@ -123,7 +119,7 @@ def dehtml(text):
def indentMultiLineText(message, n=0):
return message.replace('\n', '\n{0}'.format(' ' * n)).rstrip()
return message.replace('\n', '\n{}'.format(' ' * n)).rstrip()
def flatten_json(structure, key='', path='', flattened=None, listLimit=None):

8
src/gam/var.py
View File

@ -1076,7 +1076,7 @@ COLLABORATIVE_INBOX_ATTRIBUTES = [
'favoriteRepliesOnTop',
]
GROUP_SETTINGS_LIST_ATTRIBUTES = set([
GROUP_SETTINGS_LIST_ATTRIBUTES = {
# ACL choices
'whoCanAdd',
'whoCanApproveMembers',
@ -1117,8 +1117,8 @@ GROUP_SETTINGS_LIST_ATTRIBUTES = set([
'messageModerationLevel',
'replyTo',
'spamModerationLevel',
])
GROUP_SETTINGS_BOOLEAN_ATTRIBUTES = set([
}
GROUP_SETTINGS_BOOLEAN_ATTRIBUTES = {
'allowExternalMembers',
'allowGoogleCommunication',
'allowWebPosting',
@ -1131,7 +1131,7 @@ GROUP_SETTINGS_BOOLEAN_ATTRIBUTES = set([
'membersCanPostAsTheGroup',
'sendMessageDenyNotification',
'showInGroupDirectory',
])
}
#
# Global variables

4
src/tools/a_atleast_b.py
View File

@ -7,7 +7,7 @@ a = sys.argv[1]
b = sys.argv[2]
result = version.parse(a) >= version.parse(b)
if result:
print('OK: %s is equal or newer than %s' % (a, b))
print(f'OK: {a} is equal or newer than {b}')
else:
print('ERROR: %s is older than %s' % (a, b))
print(f'ERROR: {a} is older than {b}')
sys.exit(not result)