This repository has been archived on 2020-03-24. You can view files and clone it, but cannot push or open issues or pull requests.
quay/config_app/config_endpoints/api/__init__.py

139 lines
3.5 KiB
Python

import logging
from flask import Blueprint
from flask_restful import Resource, Api
from flask_restful.utils.cors import crossdomain
from email.utils import formatdate
from calendar import timegm
from functools import partial, wraps
from jsonschema import validate, ValidationError
from config_app.c_app import app
from config_app.config_endpoints.exception import InvalidResponse
logger = logging.getLogger(__name__)
api_bp = Blueprint('api', __name__)
CROSS_DOMAIN_HEADERS = ['Authorization', 'Content-Type', 'X-Requested-With']
class ApiExceptionHandlingApi(Api):
pass
@crossdomain(origin='*', headers=CROSS_DOMAIN_HEADERS)
def handle_error(self, error):
print('HANDLING ERROR IN API')
return super(ApiExceptionHandlingApi, self).handle_error(error)
api = ApiExceptionHandlingApi()
api.init_app(api_bp)
def format_date(date):
""" Output an RFC822 date format. """
if date is None:
return None
return formatdate(timegm(date.utctimetuple()))
def verify_not_prod(func):
@add_method_metadata('enterprise_only', True)
@wraps(func)
def wrapped(*args, **kwargs):
# Verify that we are not running on a production (i.e. hosted) stack. If so, we fail.
# This should never happen (because of the feature-flag on SUPER_USERS), but we want to be
# absolutely sure.
# if app.config['SERVER_HOSTNAME'].find('quay.io') >= 0:
# TODO(config_port) fixme
if False:
logger.error('!!! Super user method called IN PRODUCTION !!!')
raise StandardError()
return func(*args, **kwargs)
return wrapped
def resource(*urls, **kwargs):
def wrapper(api_resource):
if not api_resource:
return None
print('registering resource: ', urls)
api_resource.registered = True
api.add_resource(api_resource, *urls, **kwargs)
return api_resource
return wrapper
class ApiResource(Resource):
registered = False
method_decorators = []
def options(self):
return None, 200
def add_method_metadata(name, value):
def modifier(func):
if func is None:
return None
if '__api_metadata' not in dir(func):
func.__api_metadata = {}
func.__api_metadata[name] = value
return func
return modifier
def method_metadata(func, name):
if func is None:
return None
if '__api_metadata' in dir(func):
return func.__api_metadata.get(name, None)
return None
def no_cache(f):
@wraps(f)
def add_no_cache(*args, **kwargs):
response = f(*args, **kwargs)
if response is not None:
response.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate'
return response
return add_no_cache
def define_json_response(schema_name):
def wrapper(func):
@add_method_metadata('response_schema', schema_name)
@wraps(func)
def wrapped(self, *args, **kwargs):
schema = self.schemas[schema_name]
resp = func(self, *args, **kwargs)
if app.config['TESTING']:
try:
validate(resp, schema)
except ValidationError as ex:
raise InvalidResponse(ex.message)
return resp
return wrapped
return wrapper
nickname = partial(add_method_metadata, 'nickname')
import config_endpoints.api
import config_endpoints.api.discovery
import config_endpoints.api.suconfig
import config_endpoints.api.superuser
import config_endpoints.api.user