-
Notifications
You must be signed in to change notification settings - Fork 4
Expand file tree
/
Copy pathoptions.py
More file actions
158 lines (117 loc) · 5.45 KB
/
options.py
File metadata and controls
158 lines (117 loc) · 5.45 KB
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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
import re
from bisect import bisect
import six
from syncano.connection import ConnectionMixin
from syncano.exceptions import SyncanoValidationError, SyncanoValueError
from syncano.models.registry import registry
from syncano.utils import camelcase_to_underscore
if six.PY3:
from urllib.parse import urljoin
else:
from urlparse import urljoin
class Options(ConnectionMixin):
"""Holds metadata related to model definition."""
def __init__(self, meta=None):
self.name = None
self.plural_name = None
self.related_name = None
self.parent = None
self.parent_properties = []
self.parent_resolved = False
self.endpoints = {}
self.endpoint_fields = set()
self.fields = []
self.field_names = []
self.pk = None
if meta:
meta_attrs = meta.__dict__.copy()
for name in meta.__dict__:
if name.startswith('_') or not hasattr(self, name):
del meta_attrs[name]
for name, value in six.iteritems(meta_attrs):
setattr(self, name, value)
self.build_properties()
def build_properties(self):
for name, endpoint in six.iteritems(self.endpoints):
if 'properties' not in endpoint:
properties = self.get_path_properties(endpoint['path'])
endpoint['properties'] = properties
self.endpoint_fields.update(properties)
def contribute_to_class(self, cls, name):
if not self.name:
model_name = camelcase_to_underscore(cls.__name__)
self.name = model_name.replace('_', ' ').capitalize()
if not self.plural_name:
self.plural_name = '{0}s'.format(self.name)
if not self.related_name:
self.related_name = self.plural_name.replace(' ', '_').lower()
if self.parent and isinstance(self.parent, six.string_types):
self.parent = registry.get_model_by_name(self.parent)
self.resolve_parent_data()
setattr(cls, name, self)
def resolve_parent_data(self):
if not self.parent or self.parent_resolved:
return
parent_meta = self.parent._meta
parent_name = parent_meta.name.replace(' ', '_').lower()
parent_endpoint = parent_meta.get_endpoint('detail')
prefix = parent_endpoint['path']
for prop in parent_endpoint.get('properties', []):
if prop in parent_meta.field_names and prop not in parent_meta.parent_properties:
prop = '{0}_{1}'.format(parent_name, prop)
self.parent_properties.append(prop)
for old, new in zip(parent_endpoint['properties'], self.parent_properties):
prefix = prefix.replace(
'{{{0}}}'.format(old),
'{{{0}}}'.format(new)
)
for name, endpoint in six.iteritems(self.endpoints):
endpoint['properties'] = self.parent_properties + endpoint['properties']
endpoint['path'] = urljoin(prefix, endpoint['path'].lstrip('/'))
self.endpoint_fields.update(endpoint['properties'])
self.parent_resolved = True
def add_field(self, field):
if field.name in self.field_names:
raise SyncanoValueError('Field "{0}" already defined'.format(field.name))
self.field_names.append(field.name)
self.fields.insert(bisect(self.fields, field), field)
def get_field(self, field_name):
if not field_name:
raise SyncanoValueError('Field name is required.')
if not isinstance(field_name, six.string_types):
raise SyncanoValueError('Field name should be a string.')
for field in self.fields:
if field.name == field_name:
return field
raise SyncanoValueError('Field "{0}" not found.'.format(field_name))
def get_endpoint(self, name):
if name not in self.endpoints:
raise SyncanoValueError('Invalid path name: "{0}".'.format(name))
return self.endpoints[name]
def get_endpoint_properties(self, name):
endpoint = self.get_endpoint(name)
return endpoint['properties']
def get_endpoint_path(self, name):
endpoint = self.get_endpoint(name)
return endpoint['path']
def get_endpoint_methods(self, name):
endpoint = self.get_endpoint(name)
return endpoint['methods']
def resolve_endpoint(self, endpoint_name, properties, http_method=None):
if http_method and not self.is_http_method_available(http_method, endpoint_name):
raise SyncanoValidationError(
'HTTP method {0} not allowed for endpoint "{1}".'.format(http_method, endpoint_name)
)
endpoint = self.get_endpoint(endpoint_name)
for endpoint_name in endpoint['properties']:
if endpoint_name not in properties:
raise SyncanoValueError('Request property "{0}" is required.'.format(endpoint_name))
return endpoint['path'].format(**properties)
def is_http_method_available(self, http_method_name, endpoint_name):
available_methods = self.get_endpoint_methods(endpoint_name)
return http_method_name.lower() in available_methods
def get_endpoint_query_params(self, name, params):
properties = self.get_endpoint_properties(name)
return {k: v for k, v in six.iteritems(params) if k not in properties}
def get_path_properties(self, path):
return re.findall('/{([^}]*)}', path)