def _construct_stream_and_publisher(self, stream_name, stream_config):

        if log.isEnabledFor(logging.TRACE):  # pragma: no cover
            log.trace("%r: _construct_stream_and_publisher: "
                      "stream_name:%r, stream_config:\n%s",
                      self._platform_id, stream_name,
                      self._pp.pformat(stream_config))

        decoder = IonObjectDeserializer(obj_registry=get_obj_registry())

        if 'stream_def_dict' not in stream_config:
            # should not happen: PlatformAgent._validate_configuration validates this.
            log.error("'stream_def_dict' key not in configuration for stream %r" % stream_name)
            return

        stream_def_dict = stream_config['stream_def_dict']
        stream_def_dict['type_'] = 'StreamDefinition'
        stream_def_obj = decoder.deserialize(stream_def_dict)
        self._stream_defs[stream_name] = stream_def_obj

        routing_key           = stream_config['routing_key']
        stream_id             = stream_config['stream_id']
        exchange_point        = stream_config['exchange_point']
        parameter_dictionary  = stream_def_dict['parameter_dictionary']
        log.debug("%r: got parameter_dictionary from stream_def_dict", self._platform_id)

        self._data_streams[stream_name] = stream_id
        self._param_dicts[stream_name] = ParameterDictionary.load(parameter_dictionary)
        stream_route = StreamRoute(exchange_point=exchange_point, routing_key=routing_key)
        publisher = self._create_publisher(stream_id, stream_route)
        self._data_publishers[stream_name] = publisher

        log.debug("%r: created publisher for stream_name=%r", self._platform_id, stream_name)

    def validate_driver_configuration(self, driver_config):
        """
        Driver config must include 'oms_uri' entry.
        """
        if not 'oms_uri' in driver_config:
            log.error("'oms_uri' not present in driver_config = %s", driver_config)
            raise PlatformDriverException(msg="driver_config does not indicate 'oms_uri'")

        # validate and process ports
        if not 'ports' in driver_config:
            log.error("port information not present in driver_config = %s", driver_config)
            raise PlatformDriverException(msg="driver_config does not indicate 'ports'")

        # Create an IonObjectDeserializer
        ior = IonObjectRegistry()
        ion_deserializer = IonObjectDeserializer(obj_registry=ior)


        port_info_dict = driver_config['ports']
        for device_id, platform_port_serialized in port_info_dict.iteritems():

            platform_port = ion_deserializer.deserialize(platform_port_serialized)
            ooi_rd = OOIReferenceDesignator(platform_port.reference_designator)
            if ooi_rd.error or not ooi_rd.port:
                log.error("Invalid port information in driver_config. Reference designator: %s", platform_port.reference_designator)
            else:
                #strip leading zeros from port numbers as OMS stores as strings w/o leading zeros
                port_string = str( int(ooi_rd.port) )
                self._active_ports.append(port_string)

    def test_persisted_version(self):

        # create an initial version of SampleResource
        io_serializer = IonObjectSerializer()
        obj = IonObject('SampleResource', {'num': 9, 'other_field': 'test value'})
        obj_dict = io_serializer.serialize(obj,True)
        self.assertEquals(obj_dict['persisted_version'], 1)
        # verify that the simulated previous version does not have new_attribute
        self.assertEquals('new_attribute' in obj_dict, False)

        # simulate version increment to SampleResource that adds new_attribute
        obj_dict['type_'] = 'SampleResource_V2'
        # simulate reading the previous version after version increment
        io_deserializer = IonObjectDeserializer(obj_registry=get_obj_registry())
        obj = io_deserializer.deserialize(obj_dict)
        # verify that “new_attribute” is added and initialized with default value
        self.assertEquals(obj.new_attribute['key'], 'value')
        # verify that old attributes are still there and retain values
        self.assertEquals(obj.num, 9)
        # verify that old attributes are still there and retain values
        self.assertEquals(obj.other_field, 'test value')
        # verify that persisted_version is not updated at read
        self.assertEquals(obj_dict['persisted_version'], 1)

        # simulate update
        obj_dict = io_serializer.serialize(obj,True)
        # verify that version is updated
        self.assertEquals(obj_dict['persisted_version'], 2)

    def test_complex_version(self):

        io_serializer = IonObjectSerializer()
        obj = IonObject('SampleComplexEvent', {'num': 9, 'other_field': 'test value'})
        obj_dict = io_serializer.serialize(obj,True)
        self.assertEquals(obj_dict['persisted_version'], 1)
        # simulate a previous version data of SampleComplexEvent_V2
        obj_dict['type_'] = 'SampleComplexEvent_V2'

        # verify that the simulated previous version data has resource
        self.assertEquals('resource' in obj_dict, True)
        # verify that the simulated previous version data does not have new_attribute
        self.assertEquals('new_resource' in obj_dict, False)
        # simulate reading the previous version that does not have new_attribute
        io_deserializer = IonObjectDeserializer(obj_registry=get_obj_registry())
        obj = io_deserializer.deserialize(obj_dict)
        # verify that new attribute is added and initialized with default value
        self.assertEquals(obj.new_resource.new_attribute['key'], 'value')
        # verify that old attributes are still there
        self.assertEquals(obj.num, 9)
        # verify that old attributes are still there
        self.assertEquals(obj.other_field, 'test value')
        # verify that on read version is not yet updated
        self.assertEquals(obj_dict['persisted_version'], 1)

        # simulate create/update
        obj_dict = io_serializer.serialize(obj,True)
        # verify that version is updated
        self.assertEquals(obj_dict['persisted_version'], 2)

    def test_config(self):
        """
        test_initialize
        Test agent initialize command. This causes creation of
        driver process and transition to inactive.
        """

        # We start in uninitialized state.
        # In this state there is no driver process.
        state = self._ia_client.get_agent_state()
        self.assertEqual(state, ResourceAgentState.UNINITIALIZED)

        # Ping the agent.
        retval = self._ia_client.ping_agent()
        log.info(retval)

        # Initialize the agent.
        # The agent is spawned with a driver config, but you can pass one in
        # optinally with the initialize command. This validates the driver
        # config, launches a driver process and connects to it via messaging.
        # If successful, we switch to the inactive state.
        cmd = AgentCommand(command=ResourceAgentEvent.INITIALIZE)
        retval = self._ia_client.execute_agent(cmd)
        state = self._ia_client.get_agent_state()
        self.assertEqual(state, ResourceAgentState.INACTIVE)

        # Ping the driver proc.
        retval = self._ia_client.ping_resource()
        log.info(retval)

        decoder = IonObjectDeserializer(obj_registry=get_obj_registry())

        # Grab the alarms defined in the config.
        retval = decoder.deserialize(
            self._ia_client.get_agent(['alarms'])['alarms'])
        """
        {'status': None, 'stream_name': 'parsed', 'name': 'test_sim_warning',
        'upper_bound': 5.0, 'expr': 'x<5.0', 'upper_rel_op': '<',
        'lower_rel_op': None, 'type_': 'IntervalAlarmDef', 'value_id': 'temp',
        'lower_bound': None, 'message': 'Temperature is above test range of 5.0.',
        'current_val': None, 'type': 1}
        """
        self.assertEqual(retval[0].type_, 'IntervalAlarmDef')
        self.assertEqual(retval[0].upper_bound, 5.0)
        self.assertEqual(retval[0].expr, 'x<5.0')

        # Reset the agent. This causes the driver messaging to be stopped,
        # the driver process to end and switches us back to uninitialized.
        cmd = AgentCommand(command=ResourceAgentEvent.RESET)
        retval = self._ia_client.execute_agent(cmd)
        state = self._ia_client.get_agent_state()
        self.assertEqual(state, ResourceAgentState.UNINITIALIZED)

    def test_complex_version_del_attrib(self):

        io_serializer = IonObjectSerializer()
        # verify that extraneous fields given while creating an IonObject raises an error.
        with self.assertRaises(AttributeError):
            IonObject('SampleComplexEvent_V2', {'num': 9, 'other_field': 'test value','more_new_resource': {'key':'value'}})

        obj = IonObject('SampleComplexEvent_V2', {'num': 9, 'other_field': 'test value','new_resource': {'num': 9, 'other_field': 'test value','new_attribute':{'key':'value'}}})
        # create simulated saved data
        obj_dict = io_serializer.serialize(obj,True)
        self.assertEquals(obj_dict['persisted_version'], 2)
        # simulate a next version data of SampleComplexEvent_V2
        obj_dict['type_'] = 'SampleComplexEvent_V3'

        # verify that the simulated previous version data does have new_resource
        self.assertEquals('new_resource' in obj_dict, True)
        # note the schema version of new_resource
        self.assertEquals(obj_dict['new_resource']['persisted_version'], 2)

        # simulate reading the next version that has a new type of new_resource
        io_deserializer = IonObjectDeserializer(obj_registry=get_obj_registry())
        obj = io_deserializer.deserialize(obj_dict)

        # verify that new_resource exists
        self.assertTrue('new_resource' in obj)
        # however, verify that new_resource does not have new_attribute since type of new_resource has changed
        self.assertFalse('new_attribute' in obj.new_resource)
        # verify that the new type of new_resource has another_new_attribute that is initialized to default data
        self.assertEquals(obj.new_resource.another_new_attribute['key'], 'new_value')
        # verify on read that the schema version of new_resource replaces the old persisted_version
        self.assertEquals(obj.new_resource.persisted_version, 3)

        # verify that old attributes values of new_resource have been thrown away
        self.assertNotEquals(obj.new_resource.num, 9)
        # verify that attributes values of new_resource have been initialized to default values
        self.assertEquals(obj.new_resource.num, 0)

        # However, verify that old attributes of the resource (SampleComplexEvent) are still there
        self.assertEquals(obj.num, 9)
        # verify that old attributes are still there
        self.assertEquals(obj.other_field, 'test value')
        # verify that on read, version is not yet updated
        self.assertEquals(obj.persisted_version, 2)


        # simulate create/update
        obj_dict = io_serializer.serialize(obj,True)
        # verify that version is updated
        self.assertEquals(obj_dict['persisted_version'], 3)
        # verify that version is updated fo the subsumed object
        self.assertEquals(obj_dict['new_resource']['persisted_version'], 3)

    def test_attribute_version(self):

        io_serializer = IonObjectSerializer()

        # verify that extraneous fields given while creating an IonObject raises an error.
        with self.assertRaises(AttributeError):
            IonObject('SampleComplexEvent_V2', {'num': 9, 'other_field': 'test value','more_new_resource':
                {'key':'value'}})

        obj = IonObject('SampleComplexEvent_V2', {'num': 9, 'other_field': 'test value','new_resource':
            {'num': 9, 'other_field': 'test value','new_attribute':{'key':'value'}}})
        obj_dict = io_serializer.serialize(obj,True)
        self.assertEquals(obj_dict['persisted_version'], 2)

        # verify that the simulated previous version data does have new_resource
        self.assertEquals('new_resource' in obj_dict, True)
        # verify that the new_resource has type SampleResource_V2
        self.assertEquals(obj_dict['new_resource']['type_'],"SampleResource_V2")

        # set type to SampleComplexEvent_V3
        obj_dict['type_']="SampleComplexEvent_V3"
        obj_dict['persisted_version']=3
        # set new_resource's type to SampleResource_V3
        # so we pretend that version, not the type, of the attribute has been changed
        obj_dict['new_resource']['type_']="SampleResource_V3"

        # simulate reading SampleComplexEvent_V3 after a new version of new_resource has been introduced
        io_deserializer = IonObjectDeserializer(obj_registry=get_obj_registry())
        obj = io_deserializer.deserialize(obj_dict)

        # verify that new resource is not deleted
        self.assertTrue('new_resource' in obj)
        # verify that new resource does not have new_attribute
        self.assertFalse('new_attribute' in obj.new_resource)
        # verify that the next version of new_resource has default data in the another_new_attribute
        self.assertEquals(obj.new_resource.another_new_attribute['key'], 'new_value')
        # verify that old attributes values of new_resource have not been thrown away
        self.assertEquals(obj.new_resource.num, 9)
        # verify that values from old attributes of SampleComplexEvent_V2 are still there
        self.assertEquals(obj.num, 9)
        self.assertEquals(obj.other_field, 'test value')

        # verify that on read version is not yet updated for the subsumed object
        self.assertEquals(obj.new_resource.persisted_version, 2)

        # simulate create/update
        obj_dict = io_serializer.serialize(obj,True)
        # verify that versions are unchanged
        self.assertEquals(obj_dict['persisted_version'], 3)
        # verify that versions are updated in the subsumed object
        self.assertEquals(obj_dict['new_resource']['persisted_version'], 3)

    def __init__(self, container, datastore_name=""):
        self.container = container
        self.datastore_name = datastore_name

        # Object serialization/deserialization
        self._io_serializer = IonObjectSerializer()
        self._io_deserializer = IonObjectDeserializer(obj_registry=get_obj_registry())

class CodecInterceptor(Interceptor):
    """
    Transforms IonObject <-> dict
    """
    def __init__(self):
        Interceptor.__init__(self)
        self._io_serializer = IonObjectSerializer()
        self._io_deserializer = IonObjectDeserializer(obj_registry=get_obj_registry())

    def outgoing(self, invocation):
        #log.debug("CodecInterceptor.outgoing: %s", invocation)

        #log.debug("Payload, pre-transform: %s", invocation.message)
        invocation.message = self._io_serializer.serialize(invocation.message)
        #log.debug("Payload, post-transform: %s", invocation.message)

        return invocation

    def incoming(self, invocation):
        #log.debug("CodecInterceptor.incoming: %s", invocation)

        payload = invocation.message
        #log.debug("Payload, pre-transform: %s", payload)

        invocation.message = self._io_deserializer.deserialize(payload)
        #log.debug("Payload, post-transform: %s", invocation.message)

        return invocation

    def __init__(self, datastore_name='prototype'):
        self.datastore_name = datastore_name
        log.debug('Creating in-memory dict of dicts that will simulate data stores')
        self.root = {}

        # serializers
        self._io_serializer     = IonObjectSerializer()
        self._io_deserializer   = IonObjectDeserializer(obj_registry=get_obj_registry())

class IonSerializerDictionaryRepresentation(Representation):
    def __init__(self, id_factory):
        self.encoder = IonObjectSerializer()
        self.decoder = IonObjectDeserializer(obj_registry=get_obj_registry())
        self.id_factory = id_factory
    def encode(self, obj, add_id=False):
        out = self.encoder.serialize(obj)
        if add_id and '_id' not in out.keys():
            out['_id'] = self.id_factory.create_id()
        return out
    def decode(self, data):
        return self.decoder.deserialize(data)

    def test_version_del_attrib(self):

        io_serializer = IonObjectSerializer()

        # verify that extraneous fields given while creating an IonObject raises an error.
        with self.assertRaises(AttributeError):
            IonObject('SampleResource_V2', {'num': 9, 'other_field': 'test value','more_new_attribute': {'key':'value'}})
        # simulate creating a version 2 of SampleResource that has "new_attribute"
        obj = IonObject('SampleResource_V2', {'num': 9, 'other_field': 'test value','new_attribute': {'key':'value'}})
        obj_dict = io_serializer.serialize(obj,True)
        # verify that version is 2
        self.assertEquals(obj_dict['persisted_version'], 2)
        # verify that the simulated version 2 data does have new_attribute
        self.assertEquals('new_attribute' in obj_dict, True)


        # simulate incrementing to version 3 that does not have "new_attribute"
        obj_dict['type_'] = 'SampleResource_V3'

        # simulate reading after version increment to 3
        io_deserializer = IonObjectDeserializer(obj_registry=get_obj_registry())
        obj = io_deserializer.deserialize(obj_dict)

        # verify that new attribute is deleted
        self.assertFalse('new_attribute' in obj)
        # verify that the simulated next version data does have more_new_attribute
        self.assertEquals(obj.another_new_attribute['key'], 'new_value')

        # verify that old attributes are still there and retain their data
        self.assertEquals(obj.num, 9)
        # verify that old attributes are still there and retain their data
        self.assertEquals(obj.other_field, 'test value')
        # verify that persisted_version is not yet updated i.e. it is still 2
        self.assertEquals(obj_dict['persisted_version'], 2)

        # simulate update
        obj_dict = io_serializer.serialize(obj,True)
        # verify that version is updated
        self.assertEquals(obj_dict['persisted_version'], 3)

    def test_stream_ingestion_worker(self):
        self.start_ingestion_worker()

        context_ids, time_ctxt = self._create_param_contexts()
        pdict_id = self.dataset_management_client.create_parameter_dictionary(name='stream_ingestion_pdict', parameter_context_ids=context_ids, temporal_context='ingestion_timestamp')
        self.addCleanup(self.dataset_management_client.delete_parameter_dictionary, pdict_id)

        dataset_id = self.dataset_management_client.create_dataset(name='fake_dataset', description='fake_dataset', stream_id=self.stream_id, parameter_dictionary_id=pdict_id)
        self.addCleanup(self.dataset_management_client.delete_dataset, dataset_id)

        self.cov = self._create_coverage(dataset_id=dataset_id, parameter_dict_id=pdict_id, time_dom=self.time_dom, spatial_dom=self.spatial_dom)
        self.addCleanup(self.cov.close)

        rdt = RecordDictionaryTool(stream_definition_id=self.stream_def_id)
        rdt['conductivity'] = 1
        rdt['pressure'] = 2
        rdt['salinity'] = 3

        self.start_listener(dataset_id)

        self.publisher.publish(rdt.to_granule())
        self.data_modified = Event()
        self.data_modified.wait(30)

        cov = self.get_coverage(dataset_id)
        self.assertIsNotNone(cov.get_parameter_values('raw'))

        deserializer = IonObjectDeserializer(obj_registry=get_obj_registry())

        granule = retrieve_stream(dataset_id)
        rdt_complex = RecordDictionaryTool.load_from_granule(granule)
        rdt_complex['raw'] = [deserializer.deserialize(i) for i in rdt_complex['raw']]
        for gran in rdt_complex['raw']:
            rdt_new = RecordDictionaryTool.load_from_granule(gran)
            self.assertIn(1, rdt_new['conductivity'])
            self.assertIn(2, rdt_new['pressure'])
            self.assertIn(3, rdt_new['salinity'])
            
        cov.close()

 def _construct_streams(self, stream_info):
     decoder = IonObjectDeserializer(obj_registry=get_obj_registry())
     for (stream_name, config) in stream_info.iteritems():
         try:
             if config.has_key('stream_def_dict'):
                 stream_def_dict = config['stream_def_dict']
                 stream_def_dict['type_'] = 'StreamDefinition'
                 stream_def_obj = decoder.deserialize(stream_def_dict)
                 self._stream_defs[stream_name] = stream_def_obj
                 rdt = RecordDictionaryTool(stream_definition=stream_def_obj)
             else:
                 stream_def = config['stream_definition_ref']
                 self._stream_defs[stream_name] = stream_def
                 rdt = RecordDictionaryTool(stream_definition_id=stream_def)    
             self._agent.aparam_streams[stream_name] = rdt.fields
             self._agent.aparam_pubrate[stream_name] = 0
         except Exception as e:
             errmsg = 'Instrument agent %s' % self._agent._proc_name
             errmsg += 'error constructing stream %s. ' % stream_name
             errmsg += str(e)
             log.error(errmsg)
             
     self._agent.aparam_set_pubrate = self.aparam_set_pubrate

    def _construct_stream_and_publisher(self, stream_name, stream_config):

        # granule_publish_rate
        # records_per_granule

        if log.isEnabledFor(logging.TRACE):  # pragma: no cover
            log.trace("%r: _construct_stream_and_publisher: "
                      "stream_name:%r, stream_config:\n%s",
                      self._platform_id, stream_name,
                      self._pp.pformat(stream_config))

        decoder = IonObjectDeserializer(obj_registry=get_obj_registry())

        if 'stream_def_dict' in stream_config:
            stream_def_dict = stream_config['stream_def_dict']
            stream_def_dict['type_'] = 'StreamDefinition'
            stream_def_obj = decoder.deserialize(stream_def_dict)
            self._stream_defs[stream_name] = stream_def_obj
            log.debug("%r: using stream_def_dict", self._platform_id)

        else:  # TODO this case to be removed.
            stream_definition_ref = stream_config['stream_definition_ref']
            self._stream_defs[stream_name] = stream_definition_ref
            log.debug("%r: using stream_definition_ref", self._platform_id)

        routing_key           = stream_config['routing_key']
        stream_id             = stream_config['stream_id']
        exchange_point        = stream_config['exchange_point']
        parameter_dictionary  = stream_config['parameter_dictionary']

        self._data_streams[stream_name] = stream_id
        self._param_dicts[stream_name] = ParameterDictionary.load(parameter_dictionary)
        stream_route = StreamRoute(exchange_point=exchange_point, routing_key=routing_key)
        publisher = self._create_publisher(stream_id, stream_route)
        self._data_publishers[stream_name] = publisher

        log.debug("%r: created publisher for stream_name=%r", self._platform_id, stream_name)

    def __init__(self, datastore_name=None, config=None, scope=None, profile=None):
        """
        @param datastore_name  Name of datastore within server. May be scoped to sysname
        @param config  A server config dict with connection params
        @param scope  Prefix for the datastore name (e.g. sysname) to separate multiple systems
        """

        PostgresDataStore.__init__(self, datastore_name=datastore_name,
                                     config=config or CFG.get_safe("server.postgresql"),
                                     profile=profile or DataStore.DS_PROFILE.BASIC,
                                     scope=scope)

        # IonObject Serializers
        self._io_serializer = IonObjectSerializer()
        self._io_deserializer = IonObjectDeserializer(obj_registry=get_obj_registry())

    def test_event_version_del_attrib(self):

        io_serializer = IonObjectSerializer()

        # verify that extraneous fields given while creating an IonObject raises an error.
        with self.assertRaises(AttributeError):
            IonObject('SampleEvent_V2', {'num': 9, 'other_field': 'test value','more_new_attribute': {'key':'value'}})

        obj = IonObject('SampleEvent_V2', {'num': 9, 'other_field': 'test value','new_attribute': {'key':'value'}})
        obj_dict = io_serializer.serialize(obj,True)
        self.assertEquals(obj_dict['persisted_version'], 2)
        # simulate a next version data of SampleEvent_V2
        obj_dict['type_'] = 'SampleEvent_V3'

        # verify that the simulated previous version data does have new_attribute
        self.assertEquals('new_attribute' in obj_dict, True)
        # simulate reading the next version that does not have new_attribute
        io_deserializer = IonObjectDeserializer(obj_registry=get_obj_registry())
        obj = io_deserializer.deserialize(obj_dict)

        # verify that new attribute is deleted
        self.assertFalse('new_attribute' in obj)
        # verify that the simulated next version data does have more_new_attribute
        self.assertEquals(obj.another_new_attribute['key'], 'new_value')

        # verify that old attributes are still there
        self.assertEquals(obj.num, 9)
        # verify that old attributes are still there
        self.assertEquals(obj.other_field, 'test value')
        # verify that on read version is not yet updated
        self.assertEquals(obj_dict['persisted_version'], 2)

        # simulate create/update
        obj_dict = io_serializer.serialize(obj,True)
        # verify that version is updated
        self.assertEquals(obj_dict['persisted_version'], 3)

class CodecInterceptor(Interceptor):
    """
    Transforms IonObject <-> dict
    """
    def __init__(self):
        Interceptor.__init__(self)
        self._io_serializer = IonObjectSerializer()
        self._io_deserializer = IonObjectDeserializer(obj_registry=obj_registry)

    def outgoing(self, invocation):
        log.debug("CodecInterceptor.outgoing: %s", invocation)

        log.debug("Payload, pre-transform: %s", invocation.message)
        invocation.message = self._io_serializer.serialize(invocation.message)
        log.debug("Payload, post-transform: %s", invocation.message)

        return invocation

    def incoming(self, invocation):
        log.debug("CodecInterceptor.incoming: %s", invocation)

        payload = invocation.message
        log.debug("Payload, pre-transform: %s", payload)

        # Horrible, hacky workaround for msgpack issue
        # See http://jira.msgpack.org/browse/MSGPACK-15
        #@todo replace this with use_list in msgpack.unpackb !!!
        def convert_tuples_to_lists(obj):
            if isinstance(obj, tuple):
                res = list(obj)
                return res
            return obj

        payload = walk(payload, convert_tuples_to_lists)

        invocation.message = self._io_deserializer.deserialize(payload)
        log.debug("Payload, post-transform: %s", invocation.message)

        return invocation

class PostgresPyonDataStore(PostgresDataStore):
    """
    Base class common to both CouchDB and Couchbase datastores.
    """

    def __init__(self, datastore_name=None, config=None, scope=None, profile=None):
        """
        @param datastore_name  Name of datastore within server. May be scoped to sysname
        @param config  A server config dict with connection params
        @param scope  Prefix for the datastore name (e.g. sysname) to separate multiple systems
        """

        PostgresDataStore.__init__(self, datastore_name=datastore_name,
                                     config=config or CFG.get_safe("server.postgresql"),
                                     profile=profile or DataStore.DS_PROFILE.BASIC,
                                     scope=scope)

        # IonObject Serializers
        self._io_serializer = IonObjectSerializer()
        self._io_deserializer = IonObjectDeserializer(obj_registry=get_obj_registry())

    # -------------------------------------------------------------------------
    # Couch document operations

    def create(self, obj, object_id=None, attachments=None, datastore_name=""):
        """
        Converts ion objects to python dictionary before persisting them using the optional
        suggested identifier and creates attachments to the object.
        Returns an identifier and revision number of the object
        """
        if not isinstance(obj, IonObjectBase):
            raise BadRequest("Obj param is not instance of IonObjectBase")

        return self.create_doc(self._ion_object_to_persistence_dict(obj),
                                   object_id=object_id, datastore_name=datastore_name,
                                   attachments=attachments)

    def create_mult(self, objects, object_ids=None, allow_ids=None):
        if any([not isinstance(obj, IonObjectBase) for obj in objects]):
            raise BadRequest("Obj param is not instance of IonObjectBase")

        return self.create_doc_mult([self._ion_object_to_persistence_dict(obj) for obj in objects], object_ids)


    def update(self, obj, datastore_name=""):
        if not isinstance(obj, IonObjectBase):
            raise BadRequest("Obj param is not instance of IonObjectBase")

        return self.update_doc(self._ion_object_to_persistence_dict(obj))

    def update_mult(self, objects):
        if any([not isinstance(obj, IonObjectBase) for obj in objects]):
            raise BadRequest("Obj param is not instance of IonObjectBase")

        return self.update_doc_mult([self._ion_object_to_persistence_dict(obj) for obj in objects])


    def read(self, object_id, rev_id="", datastore_name="", object_type=None):
        if not isinstance(object_id, str):
            raise BadRequest("Object id param is not string")

        doc = self.read_doc(object_id, rev_id, datastore_name=datastore_name, object_type=object_type)
        obj = self._persistence_dict_to_ion_object(doc)

        return obj

    def read_mult(self, object_ids, datastore_name="", strict=True):
        if any([not isinstance(object_id, str) for object_id in object_ids]):
            raise BadRequest("Object ids are not string: %s" % str(object_ids))

        docs = self.read_doc_mult(object_ids, datastore_name, strict=strict)
        obj_list = [self._persistence_dict_to_ion_object(doc) if doc is not None else None for doc in docs]

        return obj_list

    def delete(self, obj, datastore_name="", object_type=None):
        if not isinstance(obj, IonObjectBase) and not isinstance(obj, str):
            raise BadRequest("Obj param is not instance of IonObjectBase or string id")
        if type(obj) is str:
            self.delete_doc(obj, datastore_name=datastore_name, object_type=object_type)
        else:
            if '_id' not in obj:
                raise BadRequest("Doc must have '_id'")
            if '_rev' not in obj:
                raise BadRequest("Doc must have '_rev'")
            self.delete_doc(self._ion_object_to_persistence_dict(obj),
                            datastore_name=datastore_name, object_type=object_type)

    def delete_mult(self, object_ids, datastore_name=None):
        return self.delete_doc_mult(object_ids, datastore_name)

    # -------------------------------------------------------------------------
    # View operations

    def find_objects_mult(self, subjects, id_only=False):
        """
        Returns a list of associations for a given list of subjects
        """
        #ds, datastore_name = self._get_datastore()
        #validate_is_instance(subjects, list, 'subjects is not a list of resource_ids')
#.........这里部分代码省略.........

class PostgresPyonDataStore(PostgresDataStore):
    """
    Base class common to both CouchDB and Couchbase datastores.
    """

    def __init__(self, datastore_name=None, config=None, scope=None, profile=None):
        """
        @param datastore_name  Name of datastore within server. May be scoped to sysname
        @param config  A server config dict with connection params
        @param scope  Prefix for the datastore name (e.g. sysname) to separate multiple systems
        """

        PostgresDataStore.__init__(self, datastore_name=datastore_name,
                                     config=config or CFG.get_safe("server.postgresql"),
                                     profile=profile or DataStore.DS_PROFILE.BASIC,
                                     scope=scope)

        # IonObject Serializers
        self._io_serializer = IonObjectSerializer()
        self._io_deserializer = IonObjectDeserializer(obj_registry=get_obj_registry())

    # -------------------------------------------------------------------------
    # Couch document operations

    def create(self, obj, object_id=None, attachments=None, datastore_name=""):
        """
        Converts ion objects to python dictionary before persisting them using the optional
        suggested identifier and creates attachments to the object.
        Returns an identifier and revision number of the object
        """
        if not isinstance(obj, IonObjectBase):
            raise BadRequest("Obj param is not instance of IonObjectBase")

        return self.create_doc(self._ion_object_to_persistence_dict(obj),
                                   object_id=object_id, datastore_name=datastore_name,
                                   attachments=attachments)

    def create_mult(self, objects, object_ids=None, allow_ids=None):
        if any([not isinstance(obj, IonObjectBase) for obj in objects]):
            raise BadRequest("Obj param is not instance of IonObjectBase")

        return self.create_doc_mult([self._ion_object_to_persistence_dict(obj) for obj in objects], object_ids)


    def update(self, obj, datastore_name=""):
        if not isinstance(obj, IonObjectBase):
            raise BadRequest("Obj param is not instance of IonObjectBase")

        return self.update_doc(self._ion_object_to_persistence_dict(obj))

    def update_mult(self, objects):
        if any([not isinstance(obj, IonObjectBase) for obj in objects]):
            raise BadRequest("Obj param is not instance of IonObjectBase")

        return self.update_doc_mult([self._ion_object_to_persistence_dict(obj) for obj in objects])


    def read(self, object_id, rev_id="", datastore_name="", object_type=None):
        if not isinstance(object_id, str):
            raise BadRequest("Object id param is not string")

        doc = self.read_doc(object_id, rev_id, datastore_name=datastore_name, object_type=object_type)
        obj = self._persistence_dict_to_ion_object(doc)

        return obj

    def read_mult(self, object_ids, datastore_name="", strict=True):
        if any([not isinstance(object_id, str) for object_id in object_ids]):
            raise BadRequest("Object ids are not string: %s" % str(object_ids))

        docs = self.read_doc_mult(object_ids, datastore_name, strict=strict)
        obj_list = [self._persistence_dict_to_ion_object(doc) if doc is not None else None for doc in docs]

        return obj_list

    def delete(self, obj, datastore_name="", object_type=None):
        if not is