Source code for io_agent
"""
@file: io_agent.py
Created on 11.12.2016 19:59
@project: GitHubRepositoryClassifier
@author: QueensGambit
The InputOutputAgent loads data (json-Data, README...) from a given repository which
is defined by strUser and strName. If the needed data has already been requested before,
then is loaded from a file. Otherwise a new connection is created.
By default the autorization of the connection is done with an API-Token
"""
from _ast import In
from clyent import json_help
from docutils.io import Input
import json
import requests
import base64
# for installtion use:
# pip install github3.py
from github3 import GitHub
from github3 import login
import os
[docs]class InputOutputAgent:
__gh = None
__bWithToken = False
__bWithTokenUpdated = False
__bRedownload = False
def __init__(self, strUser, strName):
"""
Initializes the member variables
:param strUser: username of the owner of the repository
:param strName: name of the repository
:param bWithToken: checks if a github-token shall be used for a higher api-limit
"""
self.strUser = strUser
self.strName = strName
self.strAPIUrl = "https://api.github.com/repos/" + strUser + "/" + strName
self.lstReadmePath = {"https://raw.githubusercontent.com/" + strUser + "/" + strName + "/master/README.md",
"https://raw.githubusercontent.com/" + strUser + "/" + strName + "/master/README.rst"}
@staticmethod
[docs] def setRedownload(bRedownload):
"""
sets up if the readme and json-file shall get redownload
:param bRedownload: true, false
:return:
"""
InputOutputAgent.__bRedownload = bRedownload
@staticmethod
[docs] def setWithToken(bWithToken):
"""
sets up if the github token shall be used for connection to github
:param bWithToken: true, false
:return:
"""
if bWithToken is not InputOutputAgent.__bWithToken:
# if InputOutputAgent.__gh:
# InputOutputAgent.__gh.close() # there is no .close() method
try:
InputOutputAgent.__connectToGitHub(bWithToken)
InputOutputAgent.__bWithToken = bWithToken
InputOutputAgent.__bWithTokenUpdated = True
except Exception as e:
raise e
@staticmethod
def __connectToGitHub(bWithToken):
"""
private method to establish a connection to github
:param bWithToken: true, false
:return:
"""
if InputOutputAgent.__gh is None or InputOutputAgent.__bWithTokenUpdated:
if bWithToken:
# the TokenGithubAPI is stored as an environment-variable
try:
InputOutputAgent.__gh = login(token=str(os.environ['TokenGithubAPI']))
InputOutputAgent.__bWithTokenUpdated = False
print('GithubToken is used for connection')
except Exception as ex:
raise ConnectionError('no connection to GitHub could be established')
else:
try:
InputOutputAgent.__gh = GitHub()
InputOutputAgent.__bWithTokenUpdated = False
print('No GithubToken is used for connection')
except Exception as ex:
raise ConnectionError('no connection to GitHub could be established')
# get rate limit information
rates = InputOutputAgent.__gh.rate_limit()
print('normal ratelimit info: ', rates['resources']['core']) # => your normal ratelimit info
print('search ratelimit info: ', rates['resources']['search']) # => your search ratelimit info
[docs] def loadJSONdata(self, strPathJSON):
"""
loads the requested json-data either from a file or alternatively from the web
files are exported in the './json/' directory if they were requested
"""
jsonAPI = None
# check if the json file has already been requested and was saved
if os.path.isfile(strPathJSON) and InputOutputAgent.__bRedownload is False:
# read from it
print("[INFO] Using locally cached version of repository")
with open(strPathJSON) as jsonData:
try:
if jsonData is None:
print("jsonData=None exception: ", strPathJSON)
jsonAPI = json.load(jsonData)
except Exception as ex:
raise ImportError('the json-data couldn\'t be loaded from the file: ' + strPathJSON)
raise ex
else:
InputOutputAgent.__connectToGitHub(InputOutputAgent.__bWithToken)
repo = InputOutputAgent.__gh.repository(self.strUser, self.strName)
if repo:
jsonAPI = repo.as_dict() # .as_json() returns json.dumps(obj.as_dict())
# export to json-file
with open(strPathJSON, 'w') as outfile:
json.dump(jsonAPI, outfile)
print('json-data was exported to: ', strPathJSON)
else:
raise ConnectionError('the given repository is not accessible')
return jsonAPI, self.strAPIUrl, self.lstReadmePath
[docs] def getReadme(self, strPathReadme):
"""
Gets the content from the Redme as a string.
The Readme is either loaded from file or web.
:param strPathReadme: path were the readme is loaded and exported to
:return:
"""
# Create readme directory
if not os.path.exists(strPathReadme):
os.makedirs(strPathReadme)
strPathReadme += '\\' + self.strUser + '_' + self.strName + '.txt'
# Check if readme exists already. If so, open it.
if os.path.isfile(strPathReadme) and InputOutputAgent.__bRedownload is False:
return open(strPathReadme).read()
else:
InputOutputAgent.__connectToGitHub(InputOutputAgent.__bWithToken)
repo = InputOutputAgent.__gh.repository(self.strUser, self.strName)
code64readme = repo.readme().content
# If the content of the received readme is a string and not a NullObject create
# a new file in directory. Otherwise create an empty file to prevent checking a
# repo twice.
if repo:
if isinstance(code64readme, str):
strReadme = str(base64.b64decode(code64readme))
else:
strReadme = ""
file = open(strPathReadme, "w")
file.write(strReadme)
print('readme was exported to: ', strPathReadme)
else:
raise ConnectionError('the given repository is not accessible')
return strReadme