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.