Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def wrapped(*args, **kwargs):
req_args = webargs_parser.parse({
'existing_submission_id': webargs_fields.Int(missing=None),
'cgac_code': webargs_fields.String(missing=None),
'frec_code': webargs_fields.String(missing=None)
})
# Ensure there is either an existing_submission_id, a cgac_code, or a frec_code
if req_args['existing_submission_id'] is None and req_args['cgac_code'] is None and \
req_args['frec_code'] is None:
raise ResponseException('Missing required parameter: cgac_code, frec_code, or existing_submission_id',
StatusCode.CLIENT_ERROR)
# Use codes based on existing Submission if existing_submission_id is provided, otherwise use CGAC or FREC
if req_args['existing_submission_id'] is not None:
check_existing_submission_perms(perm, req_args['existing_submission_id'])
else:
# Check permissions for the agency
if not current_user_can(perm, cgac_code=req_args['cgac_code'], frec_code=req_args['frec_code']):
raise ResponseException("User does not have permissions to write to that agency",
StatusCode.PERMISSION_DENIED)
return fn(*args, **kwargs)
'comment': fields.String(required=True),
})
def _process(self, email, comment):
self._save_report(email, comment)
return '', 204
'cross_type': webargs_fields.String(validate=webargs_validate.OneOf(FILE_TYPE_DICT.keys()))
})
def post_submission_report_url(submission, warning, file_type, cross_type):
return submission_report_url(submission, bool(warning), file_type, cross_type)
}
}
return result
externalapi_args = {
"input": fields.String(missing=None),
"apikey": fields.String(missing=None),
"t": fields.String(missing=None),
"q": fields.String(missing=None),
"query": fields.String(missing=None),
"group": fields.String(missing=None),
"limit": fields.Integer(missing=100),
"offset": fields.Integer(missing=0),
"cat": fields.String(missing=None),
"o": fields.String(missing="XML"),
"raw": fields.Integer(missing=0),
"attrs": fields.String(missing=None),
"extended": fields.Bool(missing=None),
"del": fields.String(missing=None),
"rid": fields.String(missing=None),
"genre": fields.String(missing=None),
"imdbid": fields.String(missing=None),
"tvdbid": fields.String(missing=None),
"season": fields.String(missing=None),
"ep": fields.String(missing=None),
"id": fields.String(missing=None),
"author": fields.String(missing=None),
# These aren't actually needed but the way we pass args objects along we need to have them because functions check their value
"title": fields.String(missing=None),
"category": fields.String(missing=None),
'certified': webargs_fields.String(
required=True,
validate=webargs_validate.OneOf(('mixed', 'true', 'false'))),
'sort': webargs_fields.String(missing='modified'),
'order': webargs_fields.String(missing='desc'),
'fabs': webargs_fields.Bool(missing=False),
'filters': webargs_fields.Dict(keys=webargs_fields.String(), missing={})
})
def list_submissions(certified, **kwargs):
""" List submission IDs associated with the current user """
page = kwargs.get('page')
limit = kwargs.get('limit')
sort = kwargs.get('sort')
order = kwargs.get('order')
fabs = kwargs.get('fabs')
filters = kwargs.get('filters')
return list_submissions_handler(page, limit, certified, sort, order, fabs, filters)
'filters': webargs_fields.Dict(keys=webargs_fields.String(), missing={})
})
def historic_dabs_summary(**kwargs):
filters = kwargs.get('filters')
return historic_dabs_warning_summary(filters)
'file_type': webargs_fields.String(
required=True,
validate=webargs_validate.OneOf(FILE_TYPE_DICT.keys() - {'executive_compensation', 'sub_award'})
),
'warning': webargs_fields.Bool(),
'cross_type': webargs_fields.String(validate=webargs_validate.OneOf(['program_activity', 'award_financial',
'award_procurement', 'award']))
})
def post_submission_report_url(submission, file_type, **kwargs):
warning = kwargs.get('warning')
cross_type = kwargs.get('cross_type')
return submission_report_url(submission, bool(warning), file_type, cross_type)
'start': webargs_fields.String(required=True),
'end': webargs_fields.String(required=True)
})
def generate_detached_file(file_type, cgac_code, start, end):
""" Generate a file from external API, independent from a submission """
file_manager = FileHandler(request, is_local=is_local, server_path=server_path)
return file_manager.generate_detached_file(file_type, cgac_code, start, end)
@app.route('/internalapi/getindexerstatuses')
@requires_auth("stats")
def internalapi_getindexerstatuses():
logger.debug("Get indexer statuses")
return jsonify({"indexerStatuses": get_indexer_statuses()})
internalapi__getnzbdownloads_args = {
"page": fields.Integer(missing=0),
"limit": fields.Integer(missing=100),
"type": fields.String(missing=None)
}
class SortModelSchema(Schema):
column = fields.String()
sortMode = fields.Integer()
class DownloadHistoryFilterSchema(Schema):
page = fields.Integer(missing=1)
limit = fields.Integer(missing=500)
sortModel = fields.Nested(SortModelSchema, missing=None)
@app.route('/internalapi/getnzbdownloads', methods=['POST'])
@requires_auth("stats")
@use_args(DownloadHistoryFilterSchema())
def internalapi_getnzb_downloads(args):
filterModel = request.json["filterModel"] if request.json and "filterModel" in request.json.keys() else {}
return jsonify(get_nzb_downloads(page=args["page"], limit=args["limit"], filterModel=filterModel, sortModel=args["sortModel"]))