-
Notifications
You must be signed in to change notification settings - Fork 72
/
object.py
129 lines (105 loc) · 4.4 KB
/
object.py
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
import json
from marshmallow import (
Schema,
ValidationError,
fields,
post_dump,
pre_load,
validates_schema,
)
from .attribute import AttributeItemRequestSchema, AttributeItemResponseSchema
from .metakey import MetakeyItemRequestSchema
from .tag import TagItemResponseSchema, TagRequestSchema
from .utils import UTCDateTime
class ObjectListRequestSchema(Schema):
page = fields.Int(missing=None) # legacy, to be removed in future
query = fields.Str(missing=None)
older_than = fields.Str(missing=None)
count = fields.Int(missing=10)
@validates_schema
def validate_key(self, data, **kwargs):
if data["page"] is not None and data["older_than"] is not None:
raise ValidationError(
"'page' and 'older_than' can't be used simultaneously. "
"Use 'older_than' for new code."
)
class ObjectCountRequestSchema(Schema):
query = fields.Str(missing=None)
class ObjectCreateRequestSchemaBase(Schema):
parent = fields.Str(missing=None)
metakeys = fields.Nested(MetakeyItemRequestSchema, many=True, missing=[])
attributes = fields.Nested(AttributeItemRequestSchema, many=True, missing=[])
upload_as = fields.Str(missing="*", allow_none=False)
karton_id = fields.UUID(missing=None)
karton_arguments = fields.Dict(missing={}, keys=fields.Str())
tags = fields.Nested(TagRequestSchema, many=True, missing=[])
share_3rd_party = fields.Boolean(missing=True)
class ObjectLegacyMetakeysMixin(Schema):
@pre_load
def unpack_metakeys(self, params, **kwargs):
"""
Metakeys are packed into JSON string that need to be deserialized first.
Empty string in 'metakeys' field is treated like missing key.
Request providing metakeys looks like this:
`curl ... -F="metakeys='{"metakeys": [...]}'"`
"""
params = dict(params)
if "metakeys" in params:
if params["metakeys"]:
metakeys_json = json.loads(params["metakeys"])
if "metakeys" not in metakeys_json:
raise ValidationError(
"Object provided to 'metakeys' field "
"must contain 'metakeys' key"
)
params["metakeys"] = metakeys_json["metakeys"]
else:
del params["metakeys"]
return params
class ObjectListItemResponseSchema(Schema):
id = fields.Str(attribute="dhash", required=True, allow_none=False)
type = fields.Str(required=True, allow_none=False)
tags = fields.Nested(
TagItemResponseSchema, many=True, required=True, allow_none=False
)
upload_time = UTCDateTime(required=True, allow_none=False)
class ObjectListResponseSchemaBase(Schema):
__envelope_key__ = "objects"
@post_dump(pass_many=True)
def wrap_with_envelope(self, data, many, **kwargs):
if not many:
raise ValueError("Schema supports only lists of objects")
return {self.__envelope_key__: data}
class ObjectListResponseSchema(
ObjectListResponseSchemaBase, ObjectListItemResponseSchema
):
pass
class ObjectItemResponseSchema(Schema):
id = fields.Str(attribute="dhash", required=True, allow_none=False)
type = fields.Str(required=True, allow_none=False)
tags = fields.Nested(
TagItemResponseSchema, many=True, required=True, allow_none=False
)
upload_time = UTCDateTime(required=True, allow_none=False)
favorite = fields.Boolean(required=True, allow_none=False)
parents = fields.Nested(
ObjectListItemResponseSchema, many=True, required=True, allow_none=False
)
children = fields.Nested(
ObjectListItemResponseSchema, many=True, required=True, allow_none=False
)
attributes = fields.Nested(
AttributeItemResponseSchema, many=True, required=True, allow_none=False
)
share_3rd_party = fields.Boolean(required=True, allow_none=False)
@post_dump(pass_original=True)
def get_accessible_attributes(self, data, object, **kwargs):
"""
Replace all object attributes with attributes accessible for current user
"""
object_attributes = object.get_attributes()
schema = AttributeItemResponseSchema()
attributes_serialized = schema.dump(object_attributes, many=True)
return {**data, "attributes": attributes_serialized}
class ObjectCountResponseSchema(Schema):
count = fields.Int()