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
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
|
# Copyright 2009 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Helpers to patch circular import shortcuts for the webservice.
Many of the exports for the webservice entries deliberately set various
types to `Interface` because using the real types cause circular import
problems.
The only current option is to later patch the types to the correct value.
The helper functions in this file make that easy.
"""
__metaclass__ = type
__all__ = [
'patch_choice_parameter_type',
'patch_choice_vocabulary',
'patch_collection_property',
'patch_collection_return_type',
'patch_entry_explicit_version',
'patch_entry_return_type',
'patch_list_parameter_type',
'patch_operation_explicit_version',
'patch_operations_explicit_version',
'patch_plain_parameter_type',
'patch_reference_property',
]
from lazr.restful.declarations import LAZR_WEBSERVICE_EXPORTED
from zope.schema import getFields
def patch_entry_return_type(exported_class, method_name, return_type):
"""Update return type for a webservice method that returns entries.
:param exported_class: The class containing the method.
:param method_name: The method name that you need to patch.
:param return_type: The new return type for the method.
"""
exported_class[method_name].queryTaggedValue(
LAZR_WEBSERVICE_EXPORTED)['return_type'].schema = return_type
def patch_collection_return_type(exported_class, method_name, return_type):
"""Update return type for a webservice method that returns a collection.
:param exported_class: The class containing the method.
:param method_name: The method name that you need to patch.
:param return_type: The new return type for the method.
"""
collection = exported_class[method_name].queryTaggedValue(
LAZR_WEBSERVICE_EXPORTED)
collection['return_type'].value_type.schema = return_type
def patch_list_parameter_type(exported_class, method_name, param_name,
param_type):
"""Update a list parameter type for a webservice method.
:param exported_class: The class containing the method.
:param method_name: The method name that you need to patch.
:param param_name: The name of the parameter that you need to patch.
:param param_type: The new type for the parameter.
"""
method = exported_class[method_name]
params = method.queryTaggedValue(LAZR_WEBSERVICE_EXPORTED)['params']
params[param_name].value_type = param_type
def patch_plain_parameter_type(exported_class, method_name, param_name,
param_type):
"""Update a plain parameter type for a webservice method.
:param exported_class: The class containing the method.
:param method_name: The method name that you need to patch.
:param param_name: The name of the parameter that you need to patch.
:param param_type: The new type for the parameter.
"""
exported_class[method_name].queryTaggedValue(
LAZR_WEBSERVICE_EXPORTED)['params'][param_name].schema = param_type
def patch_choice_parameter_type(exported_class, method_name, param_name,
choice_type):
"""Update a `Choice` parameter type for a webservice method.
:param exported_class: The class containing the method.
:param method_name: The method name that you need to patch.
:param param_name: The name of the parameter that you need to patch.
:param choice_type: The new choice type for the parameter.
"""
param = exported_class[method_name].queryTaggedValue(
LAZR_WEBSERVICE_EXPORTED)['params'][param_name]
param.vocabulary = choice_type
def patch_reference_property(exported_class, property_name, property_type):
"""Set the type of the given property on the given class.
:param exported_class: The class containing the property.
:param property_name: The name of the property whose type you need
to patch.
:param property_type: The new type for the property.
"""
exported_class[property_name].schema = property_type
def patch_collection_property(exported_class, property_name,
collection_type):
"""Set the collection type of the given property on the given class.
:param exported_class: The class containing the property.
:param property_name: The name of the property whose type you need
to patch.
:param collection_type: The `Collection` type.
"""
exported_class[property_name].value_type.schema = collection_type
def patch_choice_vocabulary(exported_class, method_name, param_name,
vocabulary):
"""Set the `Vocabulary` for a `Choice` parameter for a given method.
:param exported_class: The class containing the property.
:param property_name: The name of the property whose type you need
to patch.
:param vocabulary: The `Vocabulary` type.
"""
exported_class[method_name].queryTaggedValue(
LAZR_WEBSERVICE_EXPORTED)[
'params'][param_name].vocabulary = vocabulary
def patch_entry_explicit_version(interface, version):
"""Make it look as though an entry definition used as_of.
This function should be phased out in favor of actually using
as_of. This function patches the entry's fields as well as the
entry itself. Fields that are explicitly published as of a given
version (even though the entry is not) are ignored.
"""
tagged = interface.getTaggedValue(LAZR_WEBSERVICE_EXPORTED)
versioned = tagged.dict_for_name(version) or tagged.dict_for_name(None)
versioned['_as_of_was_used'] = True
# Now tag the fields.
for name, field in getFields(interface).items():
tagged = field.queryTaggedValue(LAZR_WEBSERVICE_EXPORTED)
if tagged is None:
continue
versioned = (
tagged.dict_for_name(version) or
tagged.dict_for_name(None))
if versioned is None:
# This field is explicitly published in some other version.
# Just ignore it.
continue
else:
versioned['_as_of_was_used'] = True
def patch_operations_explicit_version(interface, version, *method_names):
"""Make it look like operations' first tags were @operation_for_version.
This function should be phased out in favor of actually using
@operation_for_version, everywhere.
"""
for method in method_names:
patch_operation_explicit_version(interface, version, method)
def patch_operation_explicit_version(interface, version, method_name):
"""Make it look like an operation's first tag was @operation_for_version.
This function should be phased out in favor of actually using
@operation_for_version, everywhere.
"""
tagged = interface[method_name].getTaggedValue(LAZR_WEBSERVICE_EXPORTED)
error_prefix = "%s.%s: Attempted to patch to version %s, but " % (
interface.__name__, method_name, version)
if (len(tagged.stack) > 1
and tagged.stack[0].version == None
and tagged.stack[1].version == version):
raise ValueError(
error_prefix + (
'it is already published in %s. Did you just change '
'it to be explicitly published?' % version))
if tagged.stack[0].version == version:
raise ValueError(
error_prefix + (
'it seems to have already been patched. Does this '
'method come from a mixin used in multiple interfaces?'))
tagged.rename_version(None, version)
|