I've written a script using the Django ORM to store 'configuration settings' and interacting with the Google Drive API and another API. The script creates a base project folder and subfolders in Google Drive for a list of records, and posts links back to the source database. I'm a beginner in Python, so I'm not sure if I'm doing things the right way, or if it's completely terrible.
I think there are some areas that could be cleaned up -- for example, how I am creating the 'client' and 'service' objects repeatedly inside specific methods. I wasn't sure if it would work to create the objects and pass them between the methods though. I think I could also use the Google 'batch' and 'partial response' methods to optimize HTTP requests.
Also, I need to get this thing on a periodic task or crontab. I was interested in trying celery, so I initially set up the below jobs, and they were creating the clients, but I don't think it's working anymore in Celery.
I removed some of the script and commented areas to make it more understandable.
# Celery tasks
@periodic_task(run_every=crontab(hour="*", minute="*", day_of_week="*"))
def sync_folders():
configs = Configuration.objects.all()
for config in configs:
user = config.user
client = create_client(user, config)
try:
create_document_folders(config, client)
except Exception as inst:
pass
@periodic_task(run_every=crontab(hour="*", minute="5", day_of_week="*"))
def sync_files():
configs = Configuration.objects.all()
for config in configs:
user = config.user
client = create_client(user, config)
try:
get_updated_files(config, client)
except Exception as inst:
pass
# Rest of post is the script initiated by celery tasks above.
def create_client(user, config):
# Create the QuickBase Client object
s = config
if s is not None:
if s.qb_realm is not None:
if s.qb_realm != 'www':
baseurl = 'https://' + s.qb_realm + '.quickbase.com'
else:
baseurl = 'https://www.quickbase.com'
else:
baseurl = 'https://www.quickbase.com'
client = quickbase.Client(QUICKBASE_USER, QUICKBASE_PASSWORD, base_url=baseurl)
return client
def get_jobs_wo_folders(config, client):
if client is not None:
query = "{'" + config.root_folder_id + "'.EX.''}"
clist = [config.root_folder_id, config.root_folder_name_fid, config.root_folder_user, '3']
records = client.do_query(query, columns=clist, database=config.qb_root_dbid)
return records
def retrieve_specific_files(service, param):
# Search and retrieve a list of file resources passing a params object
result = []
page_token = None
while True:
try:
if page_token:
param['pageToken'] = page_token
files = service.files().list(**param).execute()
result.extend(files['items'])
page_token = files.get('nextPageToken')
if not page_token:
break
except errors.HttpError, error:
print 'An error occurred: %s' % error
break
return result
def get_csv(list):
si = cStringIO.StringIO()
cw = csv.writer(si)
cw.writerows(list)
return si.getvalue().strip('\r\n')
# Google helper functions
def refresh_access_token(user, config):
instance = UserSocialAuth.objects.get(user=user, provider='google-oauth2')
instance.refresh_token()
# Configuration model has a field 'last_refresh_time' that is auto-now date/time
config.save()
return instance.extra_data['access_token']
def get_access_token(config):
try:
access_token = UserSocialAuth.objects.get(user=config.user, provider='google-oauth2').extra_data['access_token']
return access_token
except Exception as inst:
pprint(inst)
def create_drive_service(config):
"""
Creates a drive service using the 'Configuration' object's
related user. Users initial signup is done with django-socialauth;
the access token and refresh token is saved with django-socialauth &
refresh token handling is done with django-socialauth.
"""
c = config
user = config.user
instance = UserSocialAuth.objects.get(user=user, provider='google-oauth2')
refreshed_at = c.token_last_refresh
now = datetime.now()
expires_in = instance.extra_data['expires']
token_age = (now - refreshed_at).seconds
if token_age > (expires_in - 120):
access_token = refresh_access_token(user, config)
else:
access_token = instance.extra_data['access_token']
try:
credentials = AccessTokenCredentials(access_token, 'Python-urllib2/2.7')
http = httplib2.Http()
http = credentials.authorize(http)
return build('drive', 'v2', http=http)
except Exception as e:
pprint(e)
# Are these next lines pointless? What should I do if the tokens consistently fail?
access_token = refresh_access_token(user)
http = httplib2.Http()
http = credentials.authorize(http)
return build('drive', 'v2', http=http)
def insert_permission(service, file_id, value, perm_type, role):
new_permission = {
'value': value,
'type': perm_type,
'role': role,
}
try:
return service.permissions().insert(
fileId=file_id, body=new_permission, sendNotificationEmails=False).execute()
except errors.HttpError, error:
print 'An error occured: %s' % error
return None
def insert_folder(config, title, parent_idNone, writer=None, mime_type='application/vnd.google-apps.folder'):
service = create_drive_service(config)
owner_permission = {
'role': 'owner',
'type': 'user',
'value': 'me',
}
body = {
'title': title,
'mimeType': mime_type,
'userPermission': owner_permission,
'fields': 'items, items/parents',
}
if parent_id:
body['parents'] = [{'id': parent_id}]
try:
folder = service.files().insert(
body=body,
).execute()
insert_permission(service, folder['id'], writer, 'user', 'writer')
return folder
except errors.HttpError, error:
print 'An error occured: %s' % error
return None
def create_document_folders(config, client):
"""
Gets records that do not have Google drive folders,
and loops through the jobs creating the necessary
base project folder and subfolders.
"""
s = config
records = get_jobs_wo_folders(config, client)
"""
A bunch of logic goes below this to loop through the list of records,
create the new folders, and append the new folder metadata to a the 'rows' list
which is later converted to records_csv
"""
if rows:
records_csv = get_csv(rows)
# Create the records in QuickBase
response = client.import_from_csv(records_csv, clist=folders_clist, database=s.folders_dbid)
