core.photovoltaicpowerstation.PhotovoltaicPowerStationCollection.on_get()

Complexity

Conditions

16

Size

Total Lines	98
Code Lines	67

Duplication

Lines	98
Ratio	100 %

Importance

Changes

0

import uuid
    @staticmethod

This code seems to be duplicated in your project.

    def on_get(req, resp):
        """
        Handle GET requests to retrieve all photovoltaic power stations

        Returns a list of all photovoltaic power stations with their complete information including:
        - Station ID, name, and UUID
        - Associated contact and cost center information
        - Station specifications and parameters
        - Related equipment and meter associations
        - SVG diagram information for visualization

        Args:
            req: Falcon request object
            resp: Falcon response object
        """
        access_control(req)
        search_query = req.get_param('q', default=None)
        if search_query is not None and len(search_query.strip()) > 0:
            search_query = search_query.strip()
        else:
            search_query = ''

        cnx = mysql.connector.connect(**config.myems_system_db)
        cursor = cnx.cursor()

        query = (" SELECT id, name, uuid "
                 " FROM tbl_contacts ")
        cursor.execute(query)
        rows_contacts = cursor.fetchall()

        contact_dict = dict()
        if rows_contacts is not None and len(rows_contacts) > 0:
            for row in rows_contacts:
                contact_dict[row[0]] = {"id": row[0],
                                        "name": row[1],
                                        "uuid": row[2]}

        query = (" SELECT id, name, uuid "
                 " FROM tbl_cost_centers ")
        cursor.execute(query)
        rows_cost_centers = cursor.fetchall()

        cost_center_dict = dict()
        if rows_cost_centers is not None and len(rows_cost_centers) > 0:
            for row in rows_cost_centers:
                cost_center_dict[row[0]] = {"id": row[0],
                                            "name": row[1],
                                            "uuid": row[2]}

        svg_dict = dict()

        query = (" SELECT id, name, uuid "
                 " FROM tbl_svgs ")
        cursor.execute(query)
        rows_svgs = cursor.fetchall()
        if rows_svgs is not None and len(rows_svgs) > 0:
            for row in rows_svgs:
                svg_dict[row[0]] = {"id": row[0],
                                    "name": row[1],
                                    "uuid": row[2]}

        query = (" SELECT id, name, uuid, "
                 "        station_code, address, latitude, longitude, rated_capacity, rated_power, "
                 "        contact_id, cost_center_id, svg_id, is_cost_data_displayed, phase_of_lifecycle, description "
                 " FROM tbl_photovoltaic_power_stations ")
        params = []
        if search_query:
            query += " WHERE name LIKE %s OR address LIKE %s OR description LIKE %s "
            params = [f'%{search_query}%', f'%{search_query}%', f'%{search_query}%']
        query += " ORDER BY id "
        cursor.execute(query, params)
        rows_photovoltaic_power_stations = cursor.fetchall()

        result = list()
        if rows_photovoltaic_power_stations is not None and len(rows_photovoltaic_power_stations) > 0:
            for row in rows_photovoltaic_power_stations:
                meta_result = {"id": row[0],
                               "name": row[1],
                               "uuid": row[2],
                               "station_code": row[3],
                               "address": row[4],
                               "latitude": row[5],
                               "longitude": row[6],
                               "rated_capacity": row[7],
                               "rated_power": row[8],
                               "contact": contact_dict.get(row[9], None),
                               "cost_center": cost_center_dict.get(row[10], None),
                               "svg": svg_dict.get(row[11], None),
                               "is_cost_data_displayed": bool(row[12]),
                               "phase_of_lifecycle": row[13],
                               "description": row[14],
                               "qrcode": 'photovoltaicpowerstation:' + row[2]}
                result.append(meta_result)

        cursor.close()
        cnx.close()
        resp.text = json.dumps(result)