Skip to content

Instantly share code, notes, and snippets.

@eflee
Last active August 4, 2017 17:18
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save eflee/12af5f66fdeb3c078171272de92b3360 to your computer and use it in GitHub Desktop.
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 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]
}
@eflee
Copy link
Author

eflee commented Aug 4, 2017

still gotta test deserializing TBH

@eflee
Copy link
Author

eflee commented Aug 4, 2017

room to optimize around line 47

@eflee
Copy link
Author

eflee commented Aug 4, 2017

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()

@eflee
Copy link
Author

eflee commented Aug 4, 2017

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"}%

@eflee
Copy link
Author

eflee commented Aug 4, 2017

@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