def _check_qos_target(self, context,
                          target_type, target_id, qos_direction):
        ret = {'router_id': None, 'port_id': None}
        if target_type is not None and target_id is not None:
            # Need to check
            try:
                if target_type == 'port':
                    target = self._core_plugin._get_port(context, target_id)
                    device_owner = target['device_owner']
                    valid_port_target = False
                    if device_owner == n_constants.DEVICE_OWNER_FLOATINGIP:
                        if qos_direction == 'egress':
                            valid_port_target = True
                    elif device_owner.startswith('compute'):
                        valid_port_target = True
                    if not valid_port_target:
                        raise ext_qos.QosInvalidPortType(
                            port_id=target_id,
                            port_type=device_owner)
                    ret['port_id'] = target_id
                elif target_type == 'router':
                    target = self._l3_plugin._get_router(context, target_id)
                    ret['router_id'] = target_id
                else:
                    # Should not reach
                    target = None
            except exc.NoResultFound:
                raise ext_qos.QosTargetNotFound(target_id=target_id)

            for qos in target.eayun_qoss:
                if qos.direction == qos_direction:
                    raise ext_qos.QosConflict()
        return ret