Skip to content

connectioncache

ConnectionCache

Cache for libraries to use with concurrent connections, processes, etc.

The cache stores the registered connections (or other objects) and allows switching between them using generated indices, user given aliases or connection objects themselves. This is useful with any library having a need for multiple concurrent connections, processes, etc.

This class is used also outside the core framework by SeleniumLibrary, SSHLibrary, etc. Backwards compatibility is thus important when doing changes.

Source code in src/robot/utils/connectioncache.py 
 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
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
class ConnectionCache:
    """Cache for libraries to use with concurrent connections, processes, etc.

    The cache stores the registered connections (or other objects) and allows
    switching between them using generated indices, user given aliases or
    connection objects themselves. This is useful with any library having a need
    for multiple concurrent connections, processes, etc.

    This class is used also outside the core framework by SeleniumLibrary,
    SSHLibrary, etc. Backwards compatibility is thus important when doing changes.
    """

    def __init__(self, no_current_msg='No open connection.'):
        self._no_current = NoConnection(no_current_msg)
        self.current = self._no_current  #: Current active connection.
        self._connections = []
        self._aliases = NormalizedDict[int]()

    @property
    def current_index(self) -> 'int|None':
        if not self:
            return None
        for index, conn in enumerate(self):
            if conn is self.current:
                return index + 1

    @current_index.setter
    def current_index(self, index: 'int|None'):
        if index is None:
            self.current = self._no_current
        else:
            self.current = self._connections[index - 1]

    def register(self, connection: Connection, alias: 'str|None' = None):
        """Registers given connection with optional alias and returns its index.

        Given connection is set to be the :attr:`current` connection.

        If alias is given, it must be a string. Aliases are case and space
        insensitive.

        The index of the first connection after initialization, and after
        :meth:`close_all` or :meth:`empty_cache`, is 1, second is 2, etc.
        """
        self.current = connection
        self._connections.append(connection)
        index = len(self._connections)
        if alias:
            self._aliases[alias] = index
        return index

    def switch(self, identifier: 'int|str|Connection') -> Connection:
        """Switches to the connection specified using the ``identifier``.

        Identifier can be an index, an alias, or a registered connection.
        Raises an error if no matching connection is found.

        Updates :attr:`current` and also returns its new value.
        """
        self.current = self.get_connection(identifier)
        return self.current

    def get_connection(self, identifier: 'int|str|Connection|None' = None) -> Connection:
        """Returns the connection specified using the ``identifier``.

        Identifier can be an index (integer or string), an alias, a registered
        connection or ``None``. If the identifier is ``None``, returns the
        current connection if it is active and raises an error if it is not.
        Raises an error also if no matching connection is found.
        """
        if identifier is None:
            if not self:
                self.current.raise_error()
            return self.current
        try:
            index = self.get_connection_index(identifier)
        except ValueError as err:
            raise RuntimeError(err.args[0])
        return self._connections[index-1]

    def get_connection_index(self, identifier: 'int|str|Connection') -> int:
        """Returns the index of the connection specified using the ``identifier``.

        Identifier can be an index (integer or string), an alias, or a registered
        connection.

        New in Robot Framework 7.0. :meth:`resolve_alias_or_index` can be used
        with earlier versions.
        """
        if isinstance(identifier, str) and identifier in self._aliases:
            return self._aliases[identifier]
        if identifier in self._connections:
            return self._connections.index(identifier) + 1
        try:
            index = int(identifier)
        except (ValueError, TypeError):
            index = -1
        if 0 < index <= len(self._connections):
            return index
        raise ValueError(f"Non-existing index or alias '{identifier}'.")

    def resolve_alias_or_index(self, alias_or_index):
        """Deprecated in RF 7.0. Use :meth:`get_connection_index` instead."""
        # This was initially added for SeleniumLibrary in RF 3.1.2.
        # https://github.com/robotframework/robotframework/issues/3125
        # The new method was added in RF 7.0. We can loudly deprecate this
        # earliest in RF 8.0.
        return self.get_connection_index(alias_or_index)

    def close_all(self, closer_method: str = 'close'):
        """Closes connections using the specified closer method and empties cache.

        If simply calling the closer method is not adequate for closing
        connections, clients should close connections themselves and use
        :meth:`empty_cache` afterward.
        """
        for conn in self._connections:
            getattr(conn, closer_method)()
        self.empty_cache()
        return self.current

    def empty_cache(self):
        """Empties the connection cache.

        Indexes of the new connections starts from 1 after this.
        """
        self.current = self._no_current
        self._connections = []
        self._aliases = NormalizedDict()

    __getitem__ = get_connection

    def __iter__(self):
        return iter(self._connections)

    def __len__(self):
        return len(self._connections)

    def __bool__(self):
        return self.current is not self._no_current

close_all(closer_method='close')

Closes connections using the specified closer method and empties cache.

If simply calling the closer method is not adequate for closing connections, clients should close connections themselves and use :meth:empty_cache afterward.

Source code in src/robot/utils/connectioncache.py 
133
134
135
136
137
138
139
140
141
142
143
def close_all(self, closer_method: str = 'close'):
    """Closes connections using the specified closer method and empties cache.

    If simply calling the closer method is not adequate for closing
    connections, clients should close connections themselves and use
    :meth:`empty_cache` afterward.
    """
    for conn in self._connections:
        getattr(conn, closer_method)()
    self.empty_cache()
    return self.current

empty_cache()

Empties the connection cache.

Indexes of the new connections starts from 1 after this.

Source code in src/robot/utils/connectioncache.py 
145
146
147
148
149
150
151
152
def empty_cache(self):
    """Empties the connection cache.

    Indexes of the new connections starts from 1 after this.
    """
    self.current = self._no_current
    self._connections = []
    self._aliases = NormalizedDict()

get_connection(identifier=None)

Returns the connection specified using the identifier.

Identifier can be an index (integer or string), an alias, a registered connection or None. If the identifier is None, returns the current connection if it is active and raises an error if it is not. Raises an error also if no matching connection is found.

Source code in src/robot/utils/connectioncache.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def get_connection(self, identifier: 'int|str|Connection|None' = None) -> Connection:
    """Returns the connection specified using the ``identifier``.

    Identifier can be an index (integer or string), an alias, a registered
    connection or ``None``. If the identifier is ``None``, returns the
    current connection if it is active and raises an error if it is not.
    Raises an error also if no matching connection is found.
    """
    if identifier is None:
        if not self:
            self.current.raise_error()
        return self.current
    try:
        index = self.get_connection_index(identifier)
    except ValueError as err:
        raise RuntimeError(err.args[0])
    return self._connections[index-1]

get_connection_index(identifier)

Returns the index of the connection specified using the identifier.

Identifier can be an index (integer or string), an alias, or a registered connection.

New in Robot Framework 7.0. :meth:resolve_alias_or_index can be used with earlier versions.

Source code in src/robot/utils/connectioncache.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def get_connection_index(self, identifier: 'int|str|Connection') -> int:
    """Returns the index of the connection specified using the ``identifier``.

    Identifier can be an index (integer or string), an alias, or a registered
    connection.

    New in Robot Framework 7.0. :meth:`resolve_alias_or_index` can be used
    with earlier versions.
    """
    if isinstance(identifier, str) and identifier in self._aliases:
        return self._aliases[identifier]
    if identifier in self._connections:
        return self._connections.index(identifier) + 1
    try:
        index = int(identifier)
    except (ValueError, TypeError):
        index = -1
    if 0 < index <= len(self._connections):
        return index
    raise ValueError(f"Non-existing index or alias '{identifier}'.")

register(connection, alias=None)

Registers given connection with optional alias and returns its index.

Given connection is set to be the :attr:current connection.

If alias is given, it must be a string. Aliases are case and space insensitive.

The index of the first connection after initialization, and after :meth:close_all or :meth:empty_cache, is 1, second is 2, etc.

Source code in src/robot/utils/connectioncache.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def register(self, connection: Connection, alias: 'str|None' = None):
    """Registers given connection with optional alias and returns its index.

    Given connection is set to be the :attr:`current` connection.

    If alias is given, it must be a string. Aliases are case and space
    insensitive.

    The index of the first connection after initialization, and after
    :meth:`close_all` or :meth:`empty_cache`, is 1, second is 2, etc.
    """
    self.current = connection
    self._connections.append(connection)
    index = len(self._connections)
    if alias:
        self._aliases[alias] = index
    return index

resolve_alias_or_index(alias_or_index)

Deprecated in RF 7.0. Use :meth:get_connection_index instead.

Source code in src/robot/utils/connectioncache.py 
125
126
127
128
129
130
131
def resolve_alias_or_index(self, alias_or_index):
    """Deprecated in RF 7.0. Use :meth:`get_connection_index` instead."""
    # This was initially added for SeleniumLibrary in RF 3.1.2.
    # https://github.com/robotframework/robotframework/issues/3125
    # The new method was added in RF 7.0. We can loudly deprecate this
    # earliest in RF 8.0.
    return self.get_connection_index(alias_or_index)

switch(identifier)

Switches to the connection specified using the identifier.

Identifier can be an index, an alias, or a registered connection. Raises an error if no matching connection is found.

Updates :attr:current and also returns its new value.

Source code in src/robot/utils/connectioncache.py 
75
76
77
78
79
80
81
82
83
84
def switch(self, identifier: 'int|str|Connection') -> Connection:
    """Switches to the connection specified using the ``identifier``.

    Identifier can be an index, an alias, or a registered connection.
    Raises an error if no matching connection is found.

    Updates :attr:`current` and also returns its new value.
    """
    self.current = self.get_connection(identifier)
    return self.current