Last active
August 4, 2017 17:18
-
-
Save eflee/12af5f66fdeb3c078171272de92b3360 to your computer and use it in GitHub Desktop.
Quick hack for json serialize/deserialize on a model with unsupported types, with pluggable interface
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
""" | |
This could probably be made a lot cleaner and add some more syntactic sugar that modifying the (de)serializers dicts on each object. First attempt | |
""" | |
class UtilModel(BaseModel): | |
class Meta: | |
abstract = True | |
class_serializers = { | |
UUID: lambda x: str(x), | |
uuid.UUID: lambda x: str(x), | |
DateTime: lambda x: x.isoformat() | |
} | |
class_deserializers = { | |
uuid.UUID: lambda x: uuid.UUID(x), | |
UUID: lambda x: UUID(x), | |
DateTime: lambda x: DateTime(x) | |
} | |
serializers = {} | |
deserializers = {} | |
@classmethod | |
def serialize(cls, name, obj): | |
if name in cls.serializers: | |
return cls.serializers[name](obj) | |
return obj | |
@classmethod | |
def deserialize(cls, name, rep): | |
if name in cls.deserializers: | |
return cls.deserializers[name](rep) | |
return rep | |
def to_json(self): | |
return json.dumps({ | |
column.model_name: self.serialize(column.model_name, | |
getattr(self, column.model_name, None)) | |
for column in self.Meta.columns | |
}) | |
@classmethod | |
def from_json(cls, json_rep): | |
json_des = {key: cls.deserialize(key, obj) for key, obj in json.loads(json_rep).items()} | |
return cls(**json_des) | |
class Rotation(UtilModel): | |
""" | |
This class represents a rotation, rotations get oncall reports | |
""" | |
id = Column(UUID, hash_key=True) | |
name = Column(String) | |
pd = Column(String) | |
serializers = { | |
"id": UtilModel.class_serializers[UUID] | |
} |
room to optimize around line 47
So, a RESTful API GET for a rotation looks like this:
@app.route('/apiv1/rotation/<uuid:rotation>', methods=["GET"]) #view
def get_rotation(rotation):
return engine.query(Rotation, key=Rotation.id == rotation).one().to_json()
and the response is this:
eflee@MacBook-Pro ~ curl -vvv http://127.0.0.1:5000/apiv1/rotation/eda7073f-b274-4931-a34e-d817fd4adc9e
* Trying 127.0.0.1...
* TCP_NODELAY set
* Connected to 127.0.0.1 (127.0.0.1) port 5000 (#0)
> GET /apiv1/rotation/eda7073f-b274-4931-a34e-d817fd4adc9e HTTP/1.1
> Host: 127.0.0.1:5000
> User-Agent: curl/7.54.0
> Accept: */*
>
* HTTP 1.0, assume close after body
< HTTP/1.0 200 OK
< Content-Type: text/html; charset=utf-8
< Content-Length: 105
< Server: Werkzeug/0.12.2 Python/3.6.2
< Date: Fri, 04 Aug 2017 16:38:08 GMT
<
* Closing connection 0
{"id": "eda7073f-b274-4931-a34e-d817fd4adc9e", "pd": "http://linkity.linkerson", "name": "TestRotation1"}%
@app.route('/apiv1/rotation/<uuid:id>', methods=["PATCH"]) #update
def patch_rotation(id):
data = request.get_json()
try:
rot = engine.query(Rotation, key=Rotation.id == id).one().to_dict()
if "id" in data:
assert data["id"]==str(id), "ID URI/Data mismatch"
rot.update(data)
engine.save(Rotation(**rot))
except ConstraintViolation:
# Rotation could not be found to patch
abort(404, "Rotation could not be found")
except AssertionError as e:
abort(400, e)
return Response(status=200)
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
still gotta test deserializing TBH