/usr/lib/python2.7/dist-packages/betamax/serializers/proxy.py is in python-betamax 0.5.1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 | # -*- coding: utf-8 -*-
from .base import BaseSerializer
import os
class SerializerProxy(BaseSerializer):
"""
This is an internal implementation detail of the betamax library.
No users implementing a serializer should be using this. Developers
working on betamax need only understand that this handles the logic
surrounding whether a cassette should be updated, overwritten, or created.
It provides one consistent way for betamax to be confident in how it
serializes the data it receives. It allows authors of Serializer classes
to not have to duplicate how files are handled. It delegates the
responsibility of actually serializing the data to those classes and
handles the rest.
"""
def __init__(self, serializer, cassette_path, allow_serialization=False):
self.proxied_serializer = serializer
self.allow_serialization = allow_serialization
self.cassette_path = cassette_path
def _ensure_path_exists(self):
if not os.path.exists(self.cassette_path):
open(self.cassette_path, 'w+').close()
@classmethod
def find(cls, serialize_with, cassette_library_dir, cassette_name):
from . import serializer_registry
serializer = serializer_registry.get(serialize_with)
if serializer is None:
raise ValueError(
'No serializer registered for {0}'.format(serialize_with)
)
cassette_path = cls.generate_cassette_name(
serializer, cassette_library_dir, cassette_name
)
return cls(serializer, cassette_path)
@staticmethod
def generate_cassette_name(serializer, cassette_library_dir,
cassette_name):
return serializer.generate_cassette_name(
cassette_library_dir, cassette_name
)
def serialize(self, cassette_data):
if not self.allow_serialization:
return
self._ensure_path_exists()
with open(self.cassette_path, 'w') as fd:
fd.write(self.proxied_serializer.serialize(cassette_data))
def deserialize(self):
self._ensure_path_exists()
data = {}
with open(self.cassette_path) as fd:
data = self.proxied_serializer.deserialize(fd.read())
return data
|