Post Method using Flask

Question

Basically I'm working on an API that receives a POST method and then uses some data to create a json and send it.

There are some constraints, for example the names that I have in my database are different of the names that need to be in the json, but the values need to be the same. So what I did was create a dictionary and associate the names in the database with the names that will be in the json.

Code Explanation:

I receive a post method with a body message that will be useful to verify if that value exists in my database. If exists I will do another query that will retrieve all the devices that belong to the value that I received in the body. And then I do another query using influxDB to get all the values in the db.

In the first part what I do is for each device that I have I check if the ID exists in the influxDB and if it does I store the value.

With that value stored I check if my device name have a correspondent (a name that I don't have in the database in the) and I round the value received before and I append to a response.

In the second part I just check what names I have in my response and then put the values that are in the dictionary but not in the response there with a value of 0 (because that are names/values that I don't have in the database but are needed in the json response).

Code:

json_data = {
            "wid":None,
            "time_loc":None,
            "date_loc":None,
            "time_utc":None,
            "temp_out": "outTemp",
            "temp_in": "inTemp",
            "hum_out": "outHumidity",
            "hum_in": "inHumidity",
            "barometer": "barometer",
            "altimeter": "altimeter",
            "raw_bar":None,
            "bar_trend":None,
            "wind_spd": "windSpeed",
            "wind_dir": "windDir",
            "wind_avg_2min":None,
            "wind_avg_10min":None,
            "wind_gust_10min":None,
            "wind_gust_dir": "windGustDir",
            "dew_point": "dewpoint",
            "dew_point_cal":None,
            "wind_chill": "windchill",
            "heat_index": "heatindex",
            "thsw_index":None,
            "uv": "UV",
            "solar_radiation":None,
            "rain_rate": "rainRate",
            "storm_rain":None,
            "rain_15min":None,
            "rain_1h":None,
            "rain_day":None,
            "rain_daily":None,
            "rain_24":None,
            "rain_month":None,
            "rain_year":None,
            "et_day": "ET" ,
            "et_month":None,
            "et_year":None,
            "temp_extra_0": "extraTemp1",
            "temp_extra_1": "extraTemp2",
            "temp_extra_2": "extraTemp3",
            "temp_extra_3": "extraTemp4",
            "temp_extra_4": "extraTemp5",
            "temp_extra_5": "extraTemp6",
            "temp_extra_6": None,
            "temp_leaf_0":"leafTemp1",
            "temp_leaf_1": "leafTemp2",
            "temp_leaf_2": None,
            "temp_leaf_3":None,
            "temp_soil_0": "soilTemp1",
            "temp_soil_1": "soilTemp2",
            "temp_soil_2": "soilTemp3",
            "temp_soil_3": "soilTemp4",
            "hum_extra_0": "extraHumid1",
            "hum_extra_1": "extraHumid2",
            "hum_extra_2": "extraHumid3",
            "hum_extra_3": "extraHumid4",
            "hum_extra_4": "extraHumid5",
            "hum_extra_5": "extraHumid6",
            "hum_extra_6": "extraHumid7",
            "soil_moist_0": "soilMoist1",
            "soil_moist_1": "soilMoist2",
            "soil_moist_2": "soilMoist3",
            "soil_moist_3": "soilMoist4",
            "wet_leaf_0": "leafWet1",
            "wet_leaf_1": "leafWet2",
            "wet_leaf_2":None,
            "wet_leaf_3":None,
            "console_batt":None,
            "stations_batt_status":None,
            "forecast_icon":None,
            "forecast_rule":None,
            "sunrise_loc_time":None,
            "sunset_loc_time":None,
    }     

@app.route("/info/", methods=["POST", "GET"])
def get_current_data_devices():
    isAuth = False
    
    auth = request.authorization
    if auth and auth.username == ApiFlask.post_auth["username"] and auth.password == ApiFlask.post_auth["password"]:
        isAuth = True
    else:
        isAuth = False
        return "Unauthorized", 401

    if isAuth:
        data = request.get_data(as_text=True)
        data = json.loads(data)
        
        g_id = data["gateway_id"]
        print(f"gateway_id: {g_id}")                
            
        valid_gateway = ApiFlask.getValidGateway(g_id)
        if(valid_gateway):

                devices = ApiFlask.getDataFromDevices(g_id)
                influx_results = influxDB.get_all_devices()
                response = []
                             
                # FIRST PART      
                for device in devices:
                    influx_value = 0
                    for key in influx_results:
                        #print(key)
                        if str(device["d_id"]) == str(key):
                            influx_value = influx_results[key]  
                            print(influx_results[key])
                            break                             
    
                    for key, val in ApiFlask.json_data.items():
                        if str(device["d_piv_slug"]) == str(val):
                            try:
                                aux_value = round(influx_value, 2)
                                #print(influx_value)
                            except:
                                aux_value = 0 
                            if str(device["d_piv_slug"]) == None: 
                                break                                       
                            response.append({key: aux_value})
                            break

                # SECOND PART    
                not_in_database = []
                for key in ApiFlask.json_data:
                    if key not in [list(item.keys())[0] for item in response]:
                        not_in_database.append({key:0})

                for item in not_in_database:
                    for key in item:
                        if response: # se a lista tiver conteudo, tem keys
                            if key not in response[0].keys():
                                response.append({key:item[key]})
                        else:
                            response.append({key:item[key]})
                
                response.sort(key=lambda x: list(ApiFlask.json_data).index(list(x.keys())[0]))
                return json.dumps(response, indent=4), 200
        else:
            return 'Bad Request' , 400

I'm wondering if this code is good, if is not very pratical... I would like to hear your opinions.

Answer 1

json_data = {
            "wid": ...

The datastructure is nice enough. But the identifier name is just terrible, it tells me very little. Call it something like db_to_json_name (if I've managed to read the code properly, still unsure of that).

---

def get_current_data_devices():
    isAuth = False

Pep-8 asks that you spell it is_auth. Consider being more verbose: is_authorized. Then there's no confusion with the notion of "is authenticated".

---

This function is, ummm, a trifle long. I'm not saying it's too long, necessarily. But it's getting there. Be on the lookout for helpers you can extract from it.

    if isAuth:

Consider bailing early with

    if not isAuth:
            return 'Bad Request' , 400

The advantage is that the rest of the function body will be four spaces closer to left margin.

---

        print(f"gateway_id: {g_id}")                

Consider using a logger here. If nothing else, you get timestamps for free.

        if(valid_gateway):

Lose the extraneous ( ) parens -- this isn't C code.

And again, you can bail early (if not valid_gateway:). That would save yet another indent level. Or you could extract the remainder as a helper function, which again would reset the indent level and would expose a test candidate for your test suite.

---

                # FIRST PART      
                ...
                # SECOND PART    

zomg, you're killing me! If "first" / "second" are references to external documentation, then include the docs with the CodeReview submission. Otherwise, I'm left asking, "What were you trying to tell me?"

My typical advice would be to break out a pair of helper functions. But usually we have a comment like # reticulate the splines so we know the helper should be named def reticulate_splines():. Here, I am utterly at a loss, you have supplied zero hints to the Gentle Reader.

                            except:
                                aux_value = 0 

No.

You seem to be coping with round(None) here. Never use a bare except, so as not to interfere with KeyboardInterrupt CTRL/C handling.

Prefer except TypeError:, if that was your intent. Or minimally catch except Exception:, as flake8 was doubtless trying to tell you when you linted.

It looks like you could sensibly extract a pair of tiny helpers just in the "first part" alone. Naming them would offer some documentation value.

---

In the "second part", you might find that set difference is more convenient than looping.

Consider renaming not_in_database to missing_from_database. Why? Humans do better with identifiers that are stated in the positive rather than the negative. It's just easier to reason about the code's behavior that way.

                        if response: 
                            if key not in response[0].keys():
                                response.append({key:item[key]})
                        else:
                            response.append({key:item[key]})

We have a pair of tests and a pair of assignments here, which seems like more code than necessary. Could testing "this AND that" give us code with just a single assignment?

                response.sort(key=lambda x: list(ApiFlask.json_data).index(list(x.keys())[0]))

This is just too much code for an anonymous one-liner. def it as a named function, please.

---

Up top you told me a pair of web verbs were valid for this function: methods=["POST", "GET"]

I am still waiting for the punchline. Does the behavior differ between those verbs? Do we have any automated unit tests to verify this function's behavior, or the behavior of its helpers? I didn't notice anything that was POST-specific.

The various DB functions seemed to read / get data. If they have side effects on tables, be sure to call that out, and consider choosing a function name which clarifies that.

---

This code achieves some of its design goals.

There is probably some review context missing from this submission. As written, without additional context, I would be unwilling to delegate maintenance tasks on this codebase and would be reluctant to accept such tasks, since some technical debt would need to be paid down prior to adding new features. There is an opportunity to improve this code prior to merging it down to main.

Answer 2

Regarding authentication: I would expect to see the logic implemented outside of your route, by using a decorator to protect the views (routes) that require authentication.

Since you're using Apiflask, then as per their documentation you would need something like:

@app.auth_required(auth)

and add some authentication code as required.

The fact that you have only one route at present does not change the fact that it's better to externalize this stuff: your function should do only one thing, and do it well. Your approach is bug-prone, since you are setting the isAuth variable at multiple places.

To reiterate what's been said before, don't swallow exceptions. Especially when you're catching all of them. Instead be specific about what kind of exception you expect to happen. Use a logger, and log them. This will help with debugging later. And if all you want is to have provide aux_value with a fallback value of 0, then it looks like a defaultdict is appropriate.