Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/flask_restx/marshalling.py: 13%

1from collections import OrderedDict 

2from functools import wraps 

3 

4from flask import request, current_app, has_app_context 

5 

6from .mask import Mask, apply as apply_mask 

7from .utils import unpack 

8 

9 

10def make(cls): 

11 if isinstance(cls, type): 

12 return cls() 

13 return cls 

14 

15 

16def marshal(data, fields, envelope=None, skip_none=False, mask=None, ordered=False): 

17 """Takes raw data (in the form of a dict, list, object) and a dict of 

18 fields to output and filters the data based on those fields. 

19 

20 :param data: the actual object(s) from which the fields are taken from 

21 :param fields: a dict of whose keys will make up the final serialized 

22 response output 

23 :param envelope: optional key that will be used to envelop the serialized 

24 response 

25 :param bool skip_none: optional key will be used to eliminate fields 

26 which value is None or the field's key not 

27 exist in data 

28 :param bool ordered: Wether or not to preserve order 

29 

30 

31 >>> from flask_restx import fields, marshal 

32 >>> data = { 'a': 100, 'b': 'foo', 'c': None } 

33 >>> mfields = { 'a': fields.Raw, 'c': fields.Raw, 'd': fields.Raw } 

34 

35 >>> marshal(data, mfields) 

36 {'a': 100, 'c': None, 'd': None} 

37 

38 >>> marshal(data, mfields, envelope='data') 

39 {'data': {'a': 100, 'c': None, 'd': None}} 

40 

41 >>> marshal(data, mfields, skip_none=True) 

42 {'a': 100} 

43 

44 >>> marshal(data, mfields, ordered=True) 

45 OrderedDict([('a', 100), ('c', None), ('d', None)]) 

46 

47 >>> marshal(data, mfields, envelope='data', ordered=True) 

48 OrderedDict([('data', OrderedDict([('a', 100), ('c', None), ('d', None)]))]) 

49 

50 >>> marshal(data, mfields, skip_none=True, ordered=True) 

51 OrderedDict([('a', 100)]) 

52 

53 """ 

54 out, has_wildcards = _marshal(data, fields, envelope, skip_none, mask, ordered) 

55 

56 if has_wildcards: 

57 # ugly local import to avoid dependency loop 

58 from .fields import Wildcard 

59 

60 items = [] 

61 keys = [] 

62 for dkey, val in fields.items(): 

63 key = dkey 

64 if isinstance(val, dict): 

65 value = marshal(data, val, skip_none=skip_none, ordered=ordered) 

66 else: 

67 field = make(val) 

68 is_wildcard = isinstance(field, Wildcard) 

69 # exclude already parsed keys from the wildcard 

70 if is_wildcard: 

71 field.reset() 

72 if keys: 

73 field.exclude |= set(keys) 

74 keys = [] 

75 value = field.output(dkey, data, ordered=ordered) 

76 if is_wildcard: 

77 

78 def _append(k, v): 

79 if skip_none and (v is None or v == OrderedDict() or v == {}): 

80 return 

81 items.append((k, v)) 

82 

83 key = field.key or dkey 

84 _append(key, value) 

85 while True: 

86 value = field.output(dkey, data, ordered=ordered) 

87 if value is None or value == field.container.format( 

88 field.default 

89 ): 

90 break 

91 key = field.key 

92 _append(key, value) 

93 continue 

94 

95 keys.append(key) 

96 if skip_none and (value is None or value == OrderedDict() or value == {}): 

97 continue 

98 items.append((key, value)) 

99 

100 items = tuple(items) 

101 

102 out = OrderedDict(items) if ordered else dict(items) 

103 

104 if envelope: 

105 out = OrderedDict([(envelope, out)]) if ordered else {envelope: out} 

106 

107 return out 

108 

109 return out 

110 

111 

112def _marshal(data, fields, envelope=None, skip_none=False, mask=None, ordered=False): 

113 """Takes raw data (in the form of a dict, list, object) and a dict of 

114 fields to output and filters the data based on those fields. 

115 

116 :param data: the actual object(s) from which the fields are taken from 

117 :param fields: a dict of whose keys will make up the final serialized 

118 response output 

119 :param envelope: optional key that will be used to envelop the serialized 

120 response 

121 :param bool skip_none: optional key will be used to eliminate fields 

122 which value is None or the field's key not 

123 exist in data 

124 :param bool ordered: Wether or not to preserve order 

125 

126 

127 >>> from flask_restx import fields, marshal 

128 >>> data = { 'a': 100, 'b': 'foo', 'c': None } 

129 >>> mfields = { 'a': fields.Raw, 'c': fields.Raw, 'd': fields.Raw } 

130 

131 >>> marshal(data, mfields) 

132 {'a': 100, 'c': None, 'd': None} 

133 

134 >>> marshal(data, mfields, envelope='data') 

135 {'data': {'a': 100, 'c': None, 'd': None}} 

136 

137 >>> marshal(data, mfields, skip_none=True) 

138 {'a': 100} 

139 

140 >>> marshal(data, mfields, ordered=True) 

141 OrderedDict([('a', 100), ('c', None), ('d', None)]) 

142 

143 >>> marshal(data, mfields, envelope='data', ordered=True) 

144 OrderedDict([('data', OrderedDict([('a', 100), ('c', None), ('d', None)]))]) 

145 

146 >>> marshal(data, mfields, skip_none=True, ordered=True) 

147 OrderedDict([('a', 100)]) 

148 

149 """ 

150 # ugly local import to avoid dependency loop 

151 from .fields import Wildcard 

152 

153 mask = mask or getattr(fields, "__mask__", None) 

154 fields = getattr(fields, "resolved", fields) 

155 if mask: 

156 fields = apply_mask(fields, mask, skip=True) 

157 

158 if isinstance(data, (list, tuple)): 

159 out = [marshal(d, fields, skip_none=skip_none, ordered=ordered) for d in data] 

160 if envelope: 

161 out = OrderedDict([(envelope, out)]) if ordered else {envelope: out} 

162 return out, False 

163 

164 has_wildcards = {"present": False} 

165 

166 def __format_field(key, val): 

167 field = make(val) 

168 if isinstance(field, Wildcard): 

169 has_wildcards["present"] = True 

170 value = field.output(key, data, ordered=ordered) 

171 return (key, value) 

172 

173 items = ( 

174 ( 

175 (k, marshal(data, v, skip_none=skip_none, ordered=ordered)) 

176 if isinstance(v, dict) 

177 else __format_field(k, v) 

178 ) 

179 for k, v in fields.items() 

180 ) 

181 

182 if skip_none: 

183 items = ( 

184 (k, v) for k, v in items if v is not None and v != OrderedDict() and v != {} 

185 ) 

186 

187 out = OrderedDict(items) if ordered else dict(items) 

188 

189 if envelope: 

190 out = OrderedDict([(envelope, out)]) if ordered else {envelope: out} 

191 

192 return out, has_wildcards["present"] 

193 

194 

195class marshal_with(object): 

196 """A decorator that apply marshalling to the return values of your methods. 

197 

198 >>> from flask_restx import fields, marshal_with 

199 >>> mfields = { 'a': fields.Raw } 

200 >>> @marshal_with(mfields) 

201 ... def get(): 

202 ... return { 'a': 100, 'b': 'foo' } 

203 ... 

204 ... 

205 >>> get() 

206 OrderedDict([('a', 100)]) 

207 

208 >>> @marshal_with(mfields, envelope='data') 

209 ... def get(): 

210 ... return { 'a': 100, 'b': 'foo' } 

211 ... 

212 ... 

213 >>> get() 

214 OrderedDict([('data', OrderedDict([('a', 100)]))]) 

215 

216 >>> mfields = { 'a': fields.Raw, 'c': fields.Raw, 'd': fields.Raw } 

217 >>> @marshal_with(mfields, skip_none=True) 

218 ... def get(): 

219 ... return { 'a': 100, 'b': 'foo', 'c': None } 

220 ... 

221 ... 

222 >>> get() 

223 OrderedDict([('a', 100)]) 

224 

225 see :meth:`flask_restx.marshal` 

226 """ 

227 

228 def __init__( 

229 self, fields, envelope=None, skip_none=False, mask=None, ordered=False 

230 ): 

231 """ 

232 :param fields: a dict of whose keys will make up the final 

233 serialized response output 

234 :param envelope: optional key that will be used to envelop the serialized 

235 response 

236 """ 

237 self.fields = fields 

238 self.envelope = envelope 

239 self.skip_none = skip_none 

240 self.ordered = ordered 

241 self.mask = Mask(mask, skip=True) 

242 

243 def __call__(self, f): 

244 @wraps(f) 

245 def wrapper(*args, **kwargs): 

246 resp = f(*args, **kwargs) 

247 mask = self.mask 

248 if has_app_context(): 

249 mask_header = current_app.config["RESTX_MASK_HEADER"] 

250 mask = request.headers.get(mask_header) or mask 

251 if isinstance(resp, tuple): 

252 data, code, headers = unpack(resp) 

253 return ( 

254 marshal( 

255 data, 

256 self.fields, 

257 self.envelope, 

258 self.skip_none, 

259 mask, 

260 self.ordered, 

261 ), 

262 code, 

263 headers, 

264 ) 

265 else: 

266 return marshal( 

267 resp, self.fields, self.envelope, self.skip_none, mask, self.ordered 

268 ) 

269 

270 return wrapper 

271 

272 

273class marshal_with_field(object): 

274 """ 

275 A decorator that formats the return values of your methods with a single field. 

276 

277 >>> from flask_restx import marshal_with_field, fields 

278 >>> @marshal_with_field(fields.List(fields.Integer)) 

279 ... def get(): 

280 ... return ['1', 2, 3.0] 

281 ... 

282 >>> get() 

283 [1, 2, 3] 

284 

285 see :meth:`flask_restx.marshal_with` 

286 """ 

287 

288 def __init__(self, field): 

289 """ 

290 :param field: a single field with which to marshal the output. 

291 """ 

292 if isinstance(field, type): 

293 self.field = field() 

294 else: 

295 self.field = field 

296 

297 def __call__(self, f): 

298 @wraps(f) 

299 def wrapper(*args, **kwargs): 

300 resp = f(*args, **kwargs) 

301 

302 if isinstance(resp, tuple): 

303 data, code, headers = unpack(resp) 

304 return self.field.format(data), code, headers 

305 return self.field.format(resp) 

306 

307 return wrapper