I'm somewhat new to python and I'm creating a script that requests
to a certain API and then uses the response
to populate a local DB
, so far I have the following:
Project Structure
├─── logs
│ ├─── app-logs.log
│ └─── err-logs.log
│
├─── secrets
│ └─── config.ini
│
├─── src
│ ├─── api
│ │ └─── send_request.py
│ ├─── interface
│ │ └─── ...
│ └───templates
│
├─── main.py
send_request.py
# send_request.py
"""
Module used to handle API requests.
"""
__version__ = '0.1'
__author__ = 'Me'
# Standard lib imports
from collections import namedtuple
# Third-party lib imports
import requests
from icecream import ic
# Class used to handle requests
class HandleRequest():
# Only for debug
ic("Starting request")
# Initial parameters
def __init__(self) -> None:
self.get_response = namedtuple('GET_Response', 'status_code response')
super().__init__()
def make_request(self, url: str, params: dict):
request = requests.Request(
method = 'GET',
url = url,
params = params
).prepare()
# Log request
ic(request)
# Get response
with requests.Session() as session:
response = session.send(request)
# Log response & status code
ic(response, response.status_code)
# Verify response status
if (response.status_code >= 200 and response.status_code <= 299):
return self.get_response(status_code = response.status_code, response = response.json())
else:
return self.get_response(status_code = response.status_code, response = None)
main.py
# main.py
__version__ = '0.1'
__author__ = 'Me'
# Third-party lib imports
import configparser
from icecream import ic
# Local application imports
from src.api.send_request import HandleRequest
# Get secrets
secrets = configparser.ConfigParser()
secrets.read('secrets/config.ini')
# Define constants
TOKEN = secrets.get('API', 'token')
URL = secrets.get('API', 'api_url')
def can_connect(table: str) -> bool:
"""
Test API connection
Parameters
----------
table : str
the name of the table you want to access/test.
Returns
-------
bool
True if response status code equals to 200 & isn't empty, False otherwise.
"""
skip = 0
params = {
"token": TOKEN,
"$skip": skip
}
handler = HandleRequest()
request = handler.make_request(URL + table, params)
if (request.status_code == 200 and len(request.response) > 0):
return True
return False
ic(can_connect('persons'))
with a valid api token/url inside config.ini
I can verify its working as intended, what I'd like to know with this post is if there's anything I could/should change in my code cause I don't really know if the way I did is the "best" way to do it.. It would be amazing for me if someone with more experience could make observations about things I should/shouldn't do.
2 Answers 2
In a pathname like logs/err-logs.log
, the -logs
suffix seems redudnant.
one level down
Based on from src.api.send_request import ...
,
it seems that main.py
belongs one level down, within src/
.
The "src" doesn't tell us anything,
since all python libraries are composed of some source code.
Of course, api
is on the vague side -- feel free to
name it foo_api
or whatever marketing name your project is using.
Similarly for vague calls like secrets.get('API', ...)
.
"""
Module used to handle API requests.
"""
Yay, we have a module-level docstring! Good.
Same critique, rephrase as "to handle FOO API...".
isort
# Standard lib imports
from ...
# Third-party lib imports
Both comments are redundant with PEP-8. Just let isort sort it for you, and such concerns disappear, you needn't give it a moment's thought.
fluff comments
# Class used to handle requests
class HandleRequest():
Ok, now we're being silly. Just delete that, it isn't shedding any new light on things, it doesn't help a maintainer, any more than this would:
i += 1 # Increment the integer index i by a certain quantity, by not less than one, and by no greater.
We write code to explain the how to human reviewers,
and write # comments
to explain the why.
clean import
ic("Starting request")
Avoid print
or other side effecting calls at import
time.
The unit test or other caller won't appreciate it.
Wait till we're within __init__
or a method.
define classes at module level
def __init__(self) -> None:
self.get_response = namedtuple('GET_Response', 'status_code response')
PEP-8 explains that this is spelled wrong. Plus, we don't need a new named tuple type for every instance of HandleRequest. Prefer:
GetResponse = namedtuple('GetResponse', 'status_code response')
Calling super (object) init isn't helping you here, may as well delete that line.
EDIT
Class definitions belong at module level. Avoid nesting one class within another. Especially avoid creating N class definitions (of a namedtuple) for N object instances.
when trying to rename self.get_response to GetResponse I get the warning "Remove the unused local variable GetResponse"
The linter is correct, it is giving you good advice.
Don't create a new local GetResponse class object and then discard it unused when __init__
exits.
Define the GetResponse class up at the module level, not nested within another class.
when trying to rename self.get_response to self.GetResponse I get the warning: "Rename this field ... to match the regular expression ^[_a-z][_a-z0-9]*$" respectively.
The linter is correct, it is giving you good advice. Class objects start with initial capital letter. We typically don't create a new class object as an object field, so the regex is rightly pointing out that you're (A.) doing something weird, and (B.) not following my original advice. "Module level" means "don't indent", not even a little. Zero indent. Please define the class at module level, in this way:
GetResponse = namedtuple("GetResponse", "status_code response")
class HandleRequest:
...
type hint FTW
def make_request(self, url: str, params: dict):
This is beautiful and I thank you for it, it is helpful.
I haven't tried running mypy
over this,
but I imagine it would be more informative if we spell out
def ... , params: dict[str, str] ...
Also, as long as we're at it, it wouldn't hurt
to ->
point out we always return a GetResponse.
redundant comment
# Log request
ic(request)
# Get response
...
# Log response & status code
...
# Verify response status
These aren't telling us anything,
as the code already stated it eloquently.
Delete.
Invent a well-named helper function if you
want to point out that we log_response_and_status_code()
.
Ok, if the issue arises again I will just bite my tongue, assuming you know it's best to elide such fluff.
raise_for_status
if (response.status_code >= 200 and response.status_code <= 299):
This is kind of OK. But conventional usage would be simply
response.raise_for_status()
and if we survive that with no fatal error, then we go on to merrily pick apart the response fields.
You made a design decision to return response=None
rather than raise the customary exception.
I've not yet read enough of the code to decide
whether bucking the trend will pay off in some way.
Certainly, no #
comment explained the underlying reasoning.
Similarly down in can_connect
.
There's more than one success status.
Plus, a 300
-series redirect is definitely not a failure.
EDIT
The thing about response=None. I did that cause I didn't know what would happen if I got an error as the response, I thought the code would just stop/exit If I tried getting the response as a json() in case of an error.
Excellent, a teaching moment!
Cool, we get to learn about Exceptions today. They're not scary "crash and burn!" things. They're actually your friend. Used wisely, they can make your code more robust, not less.
Let's look at the call site, to see how this is being used.
def can_connect( ... ) -> bool:
...
request = handler.make_request( ... )
And then we do some testing to figure out what the predicate should return.
Now, the status testing you do down in make_request
isn't
exactly wrong, it isn't so bad.
But it isn't exactly right, failing to account for some
additional "things are working!" status codes,
and it's repeated.
Worse, some of that logic is repeated up in the caller,
but in a different form.
It would be better for the "error?" decision to be made
just once. And it's convenient to delegate that decision
to the requests
library, by calling
response.raise_for_status().
Doing that would change the Public API
contract
of make_request
.
Currently it offers a contract that is on the weak side.
How can we tell?
Because caller is burdened with doing some checks before
making a potentially dangerous access of the response
field, which might be None
.
A stronger contract would say, "I will return a valid webserver response, dammit!" Now of course, not all wishes come true, and not all web requests end well. Sometimes the contract cannot be fulfilled. And that's ok. Things go wrong all the time, and not just on the web, that's life. So caller must be able to cope with some level of misery anyway.
If there's no way to successfully complete the contract,
we just call the whole thing off by raising HTTPError
.
It's as if we were never called, all bets are off.
If the caller doesn't know what to do, then by default
the exception will bubble up the call stack to main level,
where it will typically be displayed in a call trace
and then we exit.
If caller does know some appropriate way to recover
from the error, perhaps by ignoring it, or by incrementing
an error counter, then it can catch the error and deal with it.
from requests import HTTPError
def can_connect( ... ) -> bool:
...
try:
request = handler.make_request( ... )
except HTTPError:
return False
...
```
-
\$\begingroup\$ Thank you so much for pointing this things out!, I'll change/refactor my code following the tips you gave. The thing about
response=None
.. I did that cause I didn't know what would happen if I got an error as the response, I thought the code would just stop/exit If I tried getting the response as ajson()
in case of an error. \$\endgroup\$Aldimar J– Aldimar J2023年11月14日 13:13:06 +00:00Commented Nov 14, 2023 at 13:13 -
\$\begingroup\$ Another thing, when trying to rename
self.get_response
toGetResponse
orself.GetResponse
I get the following warnings: "Remove the unused local variable GetResponse" & "Rename this field ... to match the regular expression ^[_a-z][_a-z0-9]*$" respectively. \$\endgroup\$Aldimar J– Aldimar J2023年11月14日 14:04:35 +00:00Commented Nov 14, 2023 at 14:04 -
\$\begingroup\$ Ok, I understand. These details aren't always simple. I expanded my remarks a bit -- look for the pair of "EDIT" markers. \$\endgroup\$J_H– J_H2023年11月14日 16:57:50 +00:00Commented Nov 14, 2023 at 16:57
-
\$\begingroup\$ I changed almost everything following what you said, do you recommend I make another question showing the "new" code or do I "Answer" my own question with it? \$\endgroup\$Aldimar J– Aldimar J2023年11月14日 20:04:43 +00:00Commented Nov 14, 2023 at 20:04
-
\$\begingroup\$ Terrific! I would start with just an Answer, and see where it goes from there. \$\endgroup\$J_H– J_H2023年11月14日 21:56:31 +00:00Commented Nov 14, 2023 at 21:56
Here goes the new code following @J_H advices:
Project structure
├─── logs
│ ├─── app.log
│ └─── err.log
│
├─── secrets
│ └─── config.ini
│
├─── serviceName_api
│ └─── handler.py
│
├─── interface
│ └─── ...
│
├─── main.py
handler.py
# handler.py
"""
Module used to handle ServiceName API requests.
"""
__version__ = "0.1"
__author__ = "Me"
from collections import namedtuple
import requests
GetResponse = namedtuple("GetResponse", "status_code headers response")
class HandleRequest(object):
def __init__(self, url: str, params: dict[str, str]) -> None:
self.url = url
self.params = params
self.response = GetResponse
def get_response(self) -> GetResponse:
r = requests.Request(method="GET", url=self.url, params=self.params).prepare()
with requests.Session() as session:
res = session.send(r)
res.raise_for_status()
return self.response(
status_code=res.status_code, headers=res.headers, response=res.json()
)
main.py
# main.py
__version__ = "0.1"
__author__ = "Me"
import configparser
from icecream import ic
from requests import HTTPError
from serviceName_api.handler import HandleRequest
# Get secrets
secrets = configparser.ConfigParser()
secrets.read("secrets/config.ini")
# Define constants
TOKEN = secrets.get("ServiceName API", "token")
URL = secrets.get("ServiceName API", "api_url")
APP_LOG = "logs\\app.log"
ERR_LOG = "logs\\err.log"
params = {
"token": TOKEN,
"$skip": "0",
}
result = None
handler = HandleRequest(URL + "persons", params)
def can_connect() -> bool:
"""
Test API connection & set result if successful
Returns
-------
bool
True if response status code is OK, False otherwise.
If True Sets
------------
result
Request response
"""
try:
global result
result = handler.get_response()
with open(APP_LOG, 'a') as a_log:
a_log.write(f'Starting request\n'
f'Status code: {result.status_code}\n'
f'Date: {result.headers['Date']}\n'
f'Ending request\n'
f'-------------------------------------\n')
return True
except HTTPError as e:
with open(ERR_LOG, 'a') as e_log:
e_log.write(f'can_connect(): False | Request Failed\n'
f'Status code: {e.response.status_code}\n'
f'Reason: {e.response.reason}\n'
f'Date: {e.response.headers['Date']}\n'
f'Result: {result}\n'
f'-------------------------------------\n')
return False
if __name__ == "__main__":
ic(can_connect())