Python network manager class using exec()

Question

I wanted to make a small self-integrated API for my PyQt5 application.

This class is a tiny chunk of the whole project:

class NyvoNetHunterRequestManager(QObject):
    
    sending_request = pyqtSignal()
    request_sent = pyqtSignal()
    
    failed_to_send = pyqtSignal()
    
    received_valid_response = pyqtSignal()
    received_invalid_response = pyqtSignal()

    def __init__(self, url: str, data: Dict, headers: Dict, method: Literal["get", "post", "put", "delete"]):
        self.url = url
        self.data = data
        self.headers = headers
        self.method = method
        
        super(QObject, self).__init__()
        
    def fire(self) -> str:
        self.sending_request.emit()
        
        current_method_name = self.method
        
        try:
            exec(
            f"global _BACKENDRESPOND; _BACKENDRESPOND = requests.{current_method_name}(url='{self.url}', data={self.data}, headers={self.headers})"
            )
            
            global _BACKENDRESPOND
            self.response = _BACKENDRESPOND
            del _BACKENDRESPOND
            
            self.request_sent.emit()
            
        except ConnectionError:
            self.failed_to_send.emit()
            return
            
        if self.response.ok:
            self.received_valid_response.emit()
            return self.response
        
        self.received_invalid_response.emit()
        return self.response
            
        
    @property
    def url(self):
        return self._url
        
    @url.setter
    def url(self, _url):
        if not isinstance(_url, str):
            url_argument_type = type(_url).__name__
            raise ValueError(f"Expected argument type passed for the parameter( url ): ( str ) | Not ( {url_argument_type}")
        
        url_is_invalid = not is_valid_url(_url)
        if url_is_invalid:
            raise TypeError("That is not a valid url.")
            
            
        self._url = _url
        
    @property
    def data(self):
        return self._data
        
    @data.setter
    def data(self, _data):
        if not isinstance(_data, dict):
            data_argument_type = type(_data).__name__
            raise ValueError(
            f"Expected argument type passed for the parameter ( data ): ( dict ) | Not: ( {data_argument_type} )"
            )
            
            
        self._data = _data
            
    @property
    def headers(self):
        return self._headers
        
    @headers.setter
    def headers(self, _headers):
        if not isinstance(_headers, dict):
            headers_argument_type = type(_headers).__name__
            raise ValueError(f"Expected argument type passed for the parameter ( headers ): ( dict ) | Not: ( {headers_argument_type} )")
            
        
        self._headers = _headers
        
    @property
    def method(self):
        return self._method
        
    @method.setter
    def method(self, _method):
        available_http_methods = ["get", "post", "put", "delete"]
        
        if not isinstance(_method, str):
            method_argument_type = type(_method).__name__
            raise ValueError(f"Expected argument types passed for the parameter ( method ): ( str ) | Not: {method_argument_type}")
            
        if not _method in available_http_methods:
            raise TypeError("That is not a valid http method.")
            
            
        self._method = _method


module_is_runned_directly = __name__ == "__main__"
if module_is_runned_directly:
    raise DirectRunError("Database modules are not intended to run directly, they are produced for improt usage only.")

I have used the exec function because I needed the HTTP method to be passed as an argument, and I also wanted to avoid making encapsulated methods for each HTTP function.

The main idiomatic question is: Is using the exec() function in this way actually safe?

Answer 1

sending_request is not a great name for an event; similar to your other signals, it should describe something that happened i.e. request_started.

Add parameter names and types to your signals where applicable. The failure signals should receive contextual information, and the valid response signal should receive the payload (or perhaps the Response object).

Provide sane defaults for your __init__ parameters. For instance, get is overwhelmingly the most common method.

Stop using Dict. Update Python and use dict.

Instead of keeping the broken-out request parameters url, method etc. on the class, just prepare a request. Don't allow mutation of these parameters. Delete every one of your accessor methods. If you need a different request, construct a different "manager".

Don't use exec! And especially don't tie this into a non-reentrant global variable. And especially especially don't allow arbitrary code execution by directly concatenating the parameter strings.

Don't silently swallow exceptions. Re-throw them.

Delete your module_is_runned_directly; it isn't helping.

Consider upgrading from PyQt5 to PyQt6.

You should be using a requests.Session. Internally, calling any of the top-level methods like requests.get spins up a single-use session, but you can do better. There should usually be one session per domain name (or, sometimes, set of related domain names within one organization). Separately, you need a session to be able to manually send prepared requests, as I demonstrate in the first code blob.

More broadly: this is in a cursed middle ground. It's a wrapper with some events that might seem useful - but only if the method is asynchronous, and it isn't asynchronous. If you want to keep it synchronous, then... just delete everything and call requests directly. This is also over-broad boilerplate; it doesn't offer much beyond what requests already offers. It would be more useful if it spoke to the domain-specific HTTP operation, which you have not shown. (There's a decent chance that your code will never use a delete method.) But in another sense - if you wanted to keep it generic, it isn't broad enough. The list of methods isn't exhaustive; for instance there's also OPTIONS.

Since you're already drinking the Qt cool-aid, consider replacing requests with QNetworkRequest. It has plenty of signals.

Example (requests)

Covering a subset of the above,

from typing import Literal, Any

import requests
from PyQt5.QtCore import QObject, pyqtSignal, pyqtBoundSignal


class NyvoNetHunterRequestManager(QObject):
    __slots__ = 'request',

    request_started: pyqtBoundSignal = pyqtSignal()
    request_sent: pyqtBoundSignal = pyqtSignal()
    failed_to_send: pyqtBoundSignal = pyqtSignal(str, arguments=('reason',))
    received_valid_response: pyqtBoundSignal = pyqtSignal(str, arguments=('content',))
    received_invalid_response: pyqtBoundSignal = pyqtSignal(int, str, arguments=('code', 'reason'))

    def __init__(
        self,
        url: str,
        method: Literal['get', 'post', 'put', 'delete'] = 'get',
        headers: dict[str, str] | None = None,
        data: dict[str, Any] | None = None,
    ) -> None:
        self.request = requests.Request(
            method=method, url=url, headers=headers, data=data,
        ).prepare()
        super().__init__()

    def send(self, session: requests.Session) -> str:
        self.request_started.emit()

        try:
            with session.send(request=self.request, stream=True) as response:
                # Due to stream mode, will be sent before the full body is retrieved
                self.request_sent.emit()

                if response.ok:
                    self.received_valid_response.emit(response.text)
                    return response.text

                self.received_invalid_response.emit(response.status_code, response.reason)
                response.raise_for_status()
        except requests.HTTPError:
            raise
        except IOError as e:
            self.failed_to_send.emit(str(e))
            raise


def demo() -> None:
    def request_started() -> None:
        print('Request started.')
    def request_sent() -> None:
        print('Request sent.')
    def failed_to_send(reason: str) -> None:
        print('Failed to send:', reason)
    def received_valid_response(response: str) -> None:
        print('Received valid response:', response[:80], '...')
    def received_invalid_response(code: int, reason: str) -> None:
        print('Received invalid response:', code, reason)

    with requests.Session() as session:
        manager = NyvoNetHunterRequestManager(url=
            'https://google.com'  # Valid
            # 'https://invaliddomainname.wrong',  # IOError
            # 'https://google.com/invalidurlpath',  # 404
        )
        manager.request_started.connect(request_started)
        manager.request_sent.connect(request_sent)
        manager.failed_to_send.connect(failed_to_send)
        manager.received_valid_response.connect(received_valid_response)
        manager.received_invalid_response.connect(received_invalid_response)

        manager.send(session)


if __name__ == '__main__':
    demo()

A word on eval

Don't exec, don't eval. Forget that those ever existed until about 15 years into your software development career. If there were a gun to your head and you were forced to use eval for this application (which, again, you absolutely shouldn't), a safer way would look like

import requests


def request_eval(method: str, url: str) -> str:
    if not isinstance(method, str):
        raise TypeError('method must be str')
    if method.lower() not in {
        'get', 'put', 'post', 'delete',
        'head', 'options', 'connect', 'trace', 'patch',
    }:
        raise ValueError(f'Invalid method "{method}"')

    code = compile(
        source='requests.' + method + '(url=url)',
        filename='dynamic_requests_call',
        mode='eval', dont_inherit=1,
    )
    empty_globals = {'__builtins__': {}}
    req_locals = {
        'requests': requests,
        'url': url,
        # ...other request parameters
    }
    response = eval(code, empty_globals, req_locals)
    return response.text


print(
    request_eval(
        method='get', url='https://google.com',
    )[:80], '...',
)

Observe the differences. We don't allow use of any built-ins or globals, we pass in locals via variable key-value pair and not string formatting, and we ban compiler flag inheritance.

Dynamic import

For completeness, the following is a half-measure that you also shouldn't do, but is still a better idea than eval. It imports the requests library method based on name.

import requests


def request_attr(method: str, url: str) -> str:
    if not isinstance(method, str):
        raise TypeError('method must be str')
    if method.lower() not in {
        'get', 'put', 'post', 'delete',
        'head', 'options', 'connect', 'trace', 'patch',
    }:
        raise ValueError(f'Invalid method "{method}"')

    fun = getattr(requests, method)
    response = fun(url=url)
    return response.text


print(
    request_attr(
        method='get', url='https://google.com',
    )[:80], '...',
)

Answer 2

You don't need to use exec().

requests has a requests.request() function that takes the name of the method as the first argument. Use that.