File: api_decode_from_openapi.py

package info (click to toggle)
python-cloudflare 2.20.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 1,048 kB
  • sloc: python: 6,932; makefile: 138; sh: 76
file content (113 lines) | stat: -rw-r--r-- 3,664 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
""" API from OpenAPI for Cloudflare API"""

import sys
import re
import datetime
import json

API_TYPES = ['GET', 'POST', 'PATCH', 'PUT', 'DELETE']

match_identifier = re.compile(r'\{[A-Za-z0-9_\-]*\}')

def do_path(cmd, values):
    """ do_path() """

    cmds = []

    if cmd[0] != '/':
        cmd = '/' + cmd  # make sure there's a leading /

    cmd = match_identifier.sub(':id', cmd)
    if cmd[-4:] == '/:id':
        cmd = cmd[:-4]
    if cmd[-4:] == '/:id':
        cmd = cmd[:-4]

    for action in values:
        if action == '' or action.upper() not in API_TYPES:
            continue
        if 'deprecated' in values[action] and values[action]['deprecated']:
            deprecated = True
            deprecated_date = datetime.datetime.now().strftime('%Y-%m-%d')
            deprecated_already = True
        else:
            deprecated = False
            deprecated_date = ''
            deprecated_already = False

        # The requestBody/content could be one of the following:
        # "requestBody": {
        #   "content": {
        #     "application/javascript" {
        #     "application/json" {
        #     "application/octet-stream" {
        #     "application/x-ndjson" {
        #     "multipart/form-data" {

        content_type = None
        if 'requestBody' in values[action] and values[action]['requestBody']:
            request_body = values[action]['requestBody']
            if 'content' in request_body and request_body['content']:
                content_type = ','.join(list(request_body['content'].keys()))
                if content_type == 'application/json':
                    # this is the default; so we simply ignore it
                    content_type = None

        if content_type:
            v = {
                    'action': action.upper(),
                    'cmd': cmd,
                    'deprecated': deprecated,
                    'deprecated_date': deprecated_date,
                    'deprecated_already': deprecated_already,
                    'content_type': content_type
                }
        else:
            v = {
                    'action': action.upper(),
                    'cmd': cmd,
                    'deprecated': deprecated,
                    'deprecated_date': deprecated_date,
                    'deprecated_already': deprecated_already
                }
        cmds.append(v)
    return cmds

def api_decode_from_openapi(content):
    """ API decode from OpenAPI for Cloudflare API"""

    try:
        j = json.loads(content)
    except json.decoder.JSONDecodeError as e:
        raise SyntaxError('OpenAPI json decode failed: %s' % (e)) from None

    try:
        components = j['components']
        info = j['info']
        cloudflare_version = info['version']
        openapi_version = j['openapi']
        paths = j['paths']
        servers = j['servers']
    except KeyError as e:
        raise SyntaxError('OpenAPI json missing standard OpenAPI values: %s' % (e)) from None

    if len(components) == 0:
        raise SyntaxError('OpenAPI json components missing values')

    cloudflare_url = None
    for server in servers:
        try:
            cloudflare_url = server['url']
        except KeyError as e:
            pass
    if not cloudflare_url:
        raise SyntaxError('OpenAPI json servers/server missing url value')

    all_cmds = []
    for path in paths:
        if path[0] != '/':
            sys.stderr.write("OpenAPI invalid path: %s\n" % (path))
            continue
        all_cmds += do_path(path, paths[path])

    return sorted(all_cmds, key=lambda v: v['cmd']), openapi_version, cloudflare_version, cloudflare_url