"""Utilities for working with RAML files.""" from functools import reduce import re def rule_to_resource_path(rule): """Convert a flask route rule to a raml resource path. Args: rule: rule that matches a request, e.g. /product/ Returns: str: raml path, e.g. /product/{productId} """ def to_camel_case(snake_str): components = snake_str.split('_') return components[0] + ''.join(x.title() for x in components[1:]) return re.sub('<(?:[^:]*:)?([^>]*)>', lambda pat: '{' + to_camel_case(pat.group(1)) + '}', rule) def find_in_raml(api_definition, rule, method): """Find section of raml spec for route rule & method. Args: api_definition: raml spec to search rule: route rule, e.g. /product/ method: http method, e.g. "get" or "put" Returns: RamlSection """ resource_path = rule_to_resource_path(rule) resources = ['/{}'.format(s) for s in resource_path.split('/') if s != ''] resource_method = \ reduce(lambda resource, s: resource.resources[s], resources, api_definition).methods[method] return RamlSection(resource_method) def raml_header_to_schema(raml_headers): """Convert headers to json schema. Takes RAML specs for headers for an endpoint and converts to JSON Draft 3 schema to use with our validator NOTE: we have to filter out props with None as the value :( Args: raml_headers (collections.OrderedDict): the headers part of a RAML endpoint descriptor. Returns: dict: the dict representing JSON validation schema snippet. """ properties = {} for item in raml_headers: _props = raml_headers[item] props = {} for prop in _props.__dict__: if _props.__dict__[prop] is not None: props[prop] = _props.__dict__[prop] properties[item] = props schema = { '$schema': 'http://json-schema.org/draft-03/schema', 'type': 'object', 'required': True, 'properties': properties } return schema class RamlSection: """Section of a RAML api.""" def __init__(self, api_section): """Wrap raml api section.""" self.api_section = api_section def body_schema(self): """Schema for json request body.""" if not self.api_section.body: return None return self.api_section.body['application/json'].schema def header_schema(self): """Schema for validating header.""" return raml_header_to_schema(self.api_section.headers)