Return the session class for a given interface type and resource class.

        :type interface_type: constants.InterfaceType
        :type resource_class: str
        :return: Session
173166
def get_session_class(cls, interface_type, resource_class):
        
        try:
            return cls._session_classes[(interface_type, resource_class)]
        except KeyError:
            raise ValueError('No class registered for %s, %s' % (interface_type, resource_class))
76358
def interface_type(self, ift):
        
        if not isinstance(ift, str):
            ift = str(ift)
        self._attributes["if"] = ift
573224
def register(cls, interface_type, resource_class):
        
        def _internal(python_class):
            if (interface_type, resource_class) in cls._session_classes:
                logger.warning('%s is already registered in the ResourceManager. '
                               'Overwriting with %s' % ((interface_type, resource_class), python_class))

            python_class.session_type = (interface_type, resource_class)
            cls._session_classes[(interface_type, resource_class)] = python_class
            return python_class
        return _internal
292999
def _get_session(self):
        
        if self._session is None:
            session = self._session = self._database.session()
            session.create()
        return self._session
822710
def session(self):
        
        if self._session is None:
            self._session = requests.Session()
            self._session.headers.update(self._headers)
        return self._session
1221115
def get_param_type(self, index):
        """'"""

        assert index in (0, 1)

        type_info = self.type.get_param_type(index)
        type_cls = get_return_class(type_info)
        instance = type_cls(None, type_info, [], self.backend)
        instance.setup()
        return instance
768300
def get(self, data_type, options=None):
        
        if options is None:
            options = {}
        response = self._client.session.get(
            '{url}/{type}'.format(
                url=self.endpoint_url, type=data_type
            ),
            params=options
        )
        return self.process_response(response)
870107
def cast_object(self, interface_object, interface_class):
        
        name = interface_class.__name__
        i = self.manager.queryInterface(interface_object._i, name)
        return interface_class(interface=i)
420010
def get_interface_by_instance_name(self, name):
        ''''''
        with self._mutex:
            for intf in self.interfaces:
                if intf.instance_name == name:
                    return intf
            return None
566932
def remote_asset(class_obj: type) -> type:
    
    assert isinstance(class_obj, type), "class_obj is not a Class"
    global _remote_asset_resource_type
    _remote_asset_resource_type = class_obj
    return class_obj
959057
def get_type(self):
        
        if self.type_idx_value == None:
            self.type_idx_value = self.CM.get_type(self.type_idx)

        return self.type_idx_value
942278
def list_resources(self, session, query='?*::INSTR'):
        

        # For each session type, ask for the list of connected resources and
        # merge them into a single list.

        resources = sum([st.list_resources()
                         for key, st in sessions.Session.iter_valid_session_classes()], [])

        resources = rname.filter(resources, query)

        return resources
169748
def get_resources_of_type(network_id, type_id, **kwargs):
    

    nodes_with_type = db.DBSession.query(Node).join(ResourceType).filter(Node.network_id==network_id, ResourceType.type_id==type_id).all()
    links_with_type = db.DBSession.query(Link).join(ResourceType).filter(Link.network_id==network_id, ResourceType.type_id==type_id).all()
    groups_with_type = db.DBSession.query(ResourceGroup).join(ResourceType).filter(ResourceGroup.network_id==network_id, ResourceType.type_id==type_id).all()

    return nodes_with_type, links_with_type, groups_with_type
590306
def register_resources(klass, registry, resource_class):
        
        config_type = getattr(resource_class.resource_type, 'config_type', None)
        if config_type is None:
            return
        resource_class.filter_registry.register('json-diff', klass)
1262967
def interface(self, iface_name):
        
        if self.has_interface(iface_name):
            return self.interfaces[iface_name]
        else:
            raise RpcException(ERR_INVALID_PARAMS, "Unknown interface: '%s'", iface_name)
1055587
def _getResourceClass(self):
        
        if self.fromVirtualEnv:
            subcls = VirtualEnvResource
        elif os.path.isdir(self._resourcePath):
            subcls = DirectoryResource
        elif os.path.isfile(self._resourcePath):
            subcls = FileResource
        elif os.path.exists(self._resourcePath):
            raise AssertionError("Neither a file or a directory: '%s'" % self._resourcePath)
        else:
            raise AssertionError("No such file or directory: '%s'" % self._resourcePath)
        return subcls
101559
def register_subclass(cls):
    

    key = cls.interface_type, cls.resource_class

    if key in _SUBCLASSES:
        raise ValueError('Class already registered for %s and %s' % key)

    _SUBCLASSES[(cls.interface_type, cls.resource_class)] = cls

    _INTERFACE_TYPES.add(cls.interface_type)
    _RESOURCE_CLASSES[cls.interface_type].add(cls.resource_class)

    if cls.is_rc_optional:
        if cls.interface_type in _DEFAULT_RC:
            raise ValueError('Default already specified for %s' %
                             cls.interface_type)
        _DEFAULT_RC[cls.interface_type] = cls.resource_class

    return cls