Implemented first version of back-end for #1
This commit is contained in:
128
blueprints/projects.py
Normal file
128
blueprints/projects.py
Normal file
@ -0,0 +1,128 @@
|
||||
"""
|
||||
This blueprint deals with everything under /api/projects/.
|
||||
|
||||
Used environment variables:
|
||||
GITEA_URL: The URL (either intenral or external) to the Gitea instance to pull data from.
|
||||
GITEA_TOKEN: The API token for the crawler user. Only requires repository.read.
|
||||
"""
|
||||
|
||||
from urllib.parse import urlencode, urlparse, urlunparse, ParseResult
|
||||
from io import BytesIO
|
||||
import logging
|
||||
import string
|
||||
import os
|
||||
|
||||
import requests
|
||||
import flask
|
||||
import magic
|
||||
|
||||
log = logging.getLogger('blueprints.projects')
|
||||
|
||||
class Blueprint(flask.Blueprint):
|
||||
"""
|
||||
Holds the required data for this blueprint to work.
|
||||
Automatically gets the UID for the logged in user,
|
||||
and gives access to a simple constructURL() function
|
||||
to make API calls without specifying the token each time.
|
||||
"""
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
self._baseURL = urlparse(os.getenv('GITEA_URL'))
|
||||
self._apiKey = os.getenv('GITEA_TOKEN')
|
||||
self.giteaUID = None
|
||||
|
||||
r = requests.get(self.constructURL('/api/v1/user'), timeout=5)
|
||||
|
||||
if r.ok:
|
||||
self.giteaUID = r.json().get('id')
|
||||
else:
|
||||
log.error('Cannot get Gitea user ID. This blueprint will not work.')
|
||||
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
def constructURL(self, apiPath: str, query: dict=None) -> str:
|
||||
"""
|
||||
Construcs an API url from the path and queries.
|
||||
"""
|
||||
|
||||
if query is None:
|
||||
query = {}
|
||||
|
||||
query.update({'token': self._apiKey})
|
||||
|
||||
newURL = ParseResult(scheme=self._baseURL.scheme, netloc=self._baseURL.netloc,
|
||||
path=apiPath, params='', query=urlencode(query), fragment='')
|
||||
return urlunparse(newURL)
|
||||
|
||||
projectsBP = Blueprint('projectsAPI', __name__)
|
||||
|
||||
def generateStrippedProjectData(data: dict) -> dict:
|
||||
"""
|
||||
Generates the stripped down project data that the front-end expects.
|
||||
"""
|
||||
|
||||
# ----- Languages -----
|
||||
# Get the languages path and request it
|
||||
path = urlparse(data['languages_url']).path
|
||||
r = requests.get(projectsBP.constructURL(path), timeout=5)
|
||||
languages = r.json()
|
||||
# Convert all languages from bytes to percentages
|
||||
languages = {lang: size / sum(languages.values()) for lang, size in languages.items()}
|
||||
|
||||
# ----- Image -----
|
||||
if not data.get('avatar_url', '') == '':
|
||||
# Get the avatar URL
|
||||
avatarUrl = urlparse(data['avatar_url'])
|
||||
# Parse the URL to get just the hash
|
||||
imgHash = avatarUrl.path.split('/')[-1:][0]
|
||||
# Create an API path for that image
|
||||
imgPath = f'/api/projects/img/{imgHash}'
|
||||
else:
|
||||
imgPath = None
|
||||
|
||||
# ----- Git URL -----
|
||||
if not data.get('private'):
|
||||
gitUrl = data.get('html_url')
|
||||
else:
|
||||
gitUrl = None
|
||||
|
||||
newData = {}
|
||||
newData.update({'name': data.get('name')})
|
||||
newData.update({'desc': data.get('description')})
|
||||
newData.update({'issues': data.get('open_issues_count')})
|
||||
newData.update({'langs': languages})
|
||||
newData.update({'gitUrl': gitUrl})
|
||||
newData.update({'imgPath': imgPath})
|
||||
return newData
|
||||
|
||||
@projectsBP.route('/api/projects', strict_slashes=False)
|
||||
def projectsList():
|
||||
"""
|
||||
Returns a filtered list of Gitea projects.
|
||||
"""
|
||||
|
||||
# Request for all Gitea projects that the crawler account is a contributor in
|
||||
r = requests.get(projectsBP.constructURL('/api/v1/repos/search',
|
||||
{'exclusive': False, 'uid': projectsBP.giteaUID}), timeout=5)
|
||||
if r.ok:
|
||||
# Filter down every repo with generateStrippedProjectData()
|
||||
projects = [generateStrippedProjectData(data) for data in r.json().get('data')]
|
||||
return projects
|
||||
else:
|
||||
flask.abort(500)
|
||||
|
||||
@projectsBP.route('/api/projects/img/<imgHash>', strict_slashes=False)
|
||||
def img(imgHash: str):
|
||||
"""
|
||||
Forwards images from Gitea's /repo-avatars.
|
||||
"""
|
||||
|
||||
# Filter the hash
|
||||
imgHash = ''.join([c for c in imgHash if c in string.ascii_lowercase+string.digits])
|
||||
# Make the request the Gitea
|
||||
r = requests.get(projectsBP.constructURL(f'/repo-avatars/{imgHash}'), timeout=5)
|
||||
# Get the MIME type
|
||||
mime = magic.from_buffer(BytesIO(r.content).read(128), mime=True)
|
||||
# Send the image
|
||||
content = BytesIO(r.content)
|
||||
return flask.send_file(content, mime)
|
||||
Reference in New Issue
Block a user