bot.utils.datahandler._format_step() - Code Metrics - Inspection of "Search state fix" - Competitive-Cyber-Clubs/Discord-Bot - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#50)

by Cyb3r

created 2021-06-12 19:37 UTC

bot.utils.datahandler._format_step() B

↳ Parent: bot.utils.datahandler

Complexity

Conditions

Size

Total Lines	46
Code Lines	20

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	7
eloc	20
nop	1
dl	0
loc	46
rs	8
c	0
b	0
f	0

"""Handles all postgresql data and tables"""
import os
import psycopg2
import psycopg2.errors
from psycopg2.extensions import AsIs
from .tables import tables
from .logger import make_logger

# Imports the main logger
log = make_logger("database", os.getenv("LOG_LEVEL", "INFO"))

# Creates the connection to the database
connection = psycopg2.connect(os.getenv("DATABASE_URL"), sslmode="require")
cursor = connection.cursor()

DuplicateError = psycopg2.errors.lookup("23505")


def table_create() -> None:
    """Table_create

    Creates tables if they do not exist at startup. All tables are pulled from tables.py

    """
    for table in tables:
        cursor.execute(table)


def _format_step(table: str) -> str:
    """Format_step
    ---

    Returns the format string to be used in insert. This was split from insert to make it less
        complex and easier to read.

    Arguments:
    ---
        table {str} -- The name of the table for the insert

    Returns:
    ---
        str -- Returns a string that will be used for cursor execution
    """
    if table == "schools":
        query_str = (
            "INSERT INTO schools"
            "(school, region, color, id, added_by, added_by_id) "
            "VALUES (%s, %s, %s, %s, %s, %s);"
        )
    elif table == "errors":
        query_str = (
            "INSERT INTO errors"
            "(id, command, message, error, time, ack) "
            "VALUES (%s, %s, %s, %s, %s, %s);"
        )
    elif table == "reports":
        query_str = (
            "INSERT INTO reports"
            "(id, name, name_id, message, time) "
            "VALUES (%s, %s, %s, %s, %s);"
        )
    elif table == "admin_channels":
        query_str = (
            "INSERT INTO admin_channels (name, id, log) "
            "VALUES (%s, %s, %s) ON CONFLICT DO NOTHING;"
        )
    elif table == "bot_admins":
        query_str = "INSERT INTO bot_admins (name, id) VALUES (%s, %s) ON CONFLICT DO NOTHING;"
    elif table == "regions":
        query_str = "INSERT INTO regions (name, id) VALUES (%s, %s)"
    else:
        log.error("Table {} not found.".format(table))
        return "error"
    return query_str


def _result_parser(column: str, fetched: list) -> list:
    """Result_parser
    ---

    Arguments:
    ---
        column {str} -- The name of the column(s) only used to determine how to parse the results.
        fetched {list} -- The results pulled from the table.

    Returns:
    ---
        list -- A normal list or a list of tuples.
    """
    # fetched is a list so it does not need breaking up
    if column == "*" or column.find(",") != -1:
        result = fetched
    # Breaks up the tuples to a standard list.
    else:
        result = [x[0] for x in fetched]
    return result


async def insert(table: str, data: list) -> [None, str]:
    """Insert
    ---

    Asynchronous Function

    Inserts a new row to an existing table. Get the string to execute with from :ref:`_format_step`.

    Arguments:
    ---
        table {str} -- c to perform the insert on
        data {list} -- The data that gets placed into the format_str

    Returns:
    ---
        str -- In the event of an error inserting into the table the string 'error' will be
            returned. If there is no error then 'None' will be returned.
    PostgreSQL Equivalent:
    ---
    INSERT into :ref:`table` VALUE (:ref:`*data`);
    """
    format_str = _format_step(table)
    if format_str == "error":
        return "error"
    log.debug("String: {} Data {}".format(format_str, " ".join(map(str, data))))
    try:
        # Tables with 6 values
        if table in ["schools", "errors"]:
            cursor.execute(format_str, (data[0], data[1], data[2], data[3], data[4], data[5]))
        # Tables with 5 values
        elif table in ["reports"]:
            cursor.execute(format_str, (data[0], data[1], data[2], data[3], data[4]))
        # Tables with 3 values
        elif table == "admin_channels":
            cursor.execute(format_str, (data[0], data[1], data[2]))
        # Tables with 2 values
        elif table in ["bot_admins", "regions"]:
            cursor.execute(format_str, (data[0], data[1]))
        connection.commit()
        return None
    except psycopg2.Error as pge:
        log.error(pge)
        cursor.rollback()
        if isinstance(pge, DuplicateError):
            return "duplicate"
        return "error"


async def fetch(table: str, column: str) -> list:
    """Fetch
    ---

    Asynchronous Function

    Retrieves values from the :ref:`column` from the :ref:`table`.

    Arguments:
    ---
        table {str} -- Name of the table that data is being fetched from.
        column {str} -- The column(s) that is being fetched. Multiple columns need to comma
            separated, if all columns are wanted then use '*'.

    Returns:
    ---
        list -- A list of tuples for '*' or multiple columns. For one column it is a list.

    Postgresql Equivalent:
    ---
    SELECT :ref:`column` from :ref:`table`;
    """
    try:
        format_str = "SELECT %s FROM %s;"
        cursor.execute(format_str, (AsIs(column), AsIs(table)))
        fetched = cursor.fetchall()
        return _result_parser(column, fetched)
    except psycopg2.Error as pge:
        log.error(pge)


async def select(
    table: str,
    column: str,
    where_column: str,
    where_value: [str, bool, int],
    symbol: [str, bool] = "=",
) -> list:
    """Select
    ---

    Asynchronous Function

    Selects one row from the table based on selector.


    Arguments:
    ---
        table {str} -- Name of the table that data is being fetched from.
        column {str} -- The column(s) that is being fetched. Multiple columns need to comma
            separated, if all columns are wanted then use '*'.
        where_column {str} -- The column that is going have the value of :ref:`where_value`.
        where_value {str} -- The value that you are matching.
        symbol {str} -- [description] (default: {"="})

    Returns:
    ---
        list -- List of values that are the results.

    Postgresql Equivalent:
    ---
    SELECT :ref:`column` FROM :ref:`table` WHERE :ref:`where_column` :ref:`symbol` :ref:`where_value`;  # noqa: E501 pylint: disable=line-too-long
    """
    try:
        format_str = "SELECT %s FROM %s WHERE %s %s %s;"
        cursor.execute(
            format_str,
            (AsIs(column), AsIs(table), AsIs(where_column), AsIs(symbol), where_value),
        )
        fetched = cursor.fetchall()
        return _result_parser(column, fetched)
    except psycopg2.Error as pge:
        log.error(pge)
        cursor.execute("ROLLBACK")


async def update(
    table: str, column: str, where_value: str, new_value: [str, bool], where_column: str = ""
) -> None:
    """Update
    ---

    Asynchronous Function

    Updates a value in the table

    Arguments:
        table {str} -- Name of the table that the data is being updated on.
        column {str} -- The column that is being updated. Multiple columns are not supported.
        where_value {str, bool} -- The value that is going to be updated.
        new_value {str} -- The new value for :ref:`where_value`

    Postgresql Equivalent:
    ---
    UPDATE :ref:`table` SET :ref:`column` = :ref:`new_value`
    WHERE :ref:`check_column` = :ref:`where_value`;
    """
    if not where_column:
        where_column = column
    try:
        format_str = "UPDATE %s SET %s = %s where %s = %s"
        cursor.execute(
            format_str,
            (AsIs(table), AsIs(column), new_value, AsIs(where_column), where_value),
        )
        connection.commit()
    except psycopg2.Error as pge:
        log.error(pge)
        cursor.execute("ROLLBACK")


async def delete(table: str, column: str, value: str) -> None:
    """Delete
    ---
    Asynchronous Function

    Removes an entry from the table where :ref:`value` is equal.

    Arguments:
    ---
        table {str} -- Name of the table that the data is being deleted from.

        column {str} -- The column to which the :ref:`value` is going to match.

        value {str} -- The value in the row which is going to match to a value in :ref:`column`

    Postgresql Equivalent:
    ---
    DELETE FROM :ref:`table` WHERE :ref:`column` = :ref:`value`;
    """
    try:
        log.info("Deleting {} where {} from {}".format(column, value, table))
        format_str = "DELETE FROM %s WHERE %s = %s"
        cursor.execute(format_str, (AsIs(table), AsIs(column), value))
        connection.commit()
    except psycopg2.Error as pge:
        log.error(pge)
        cursor.execute("ROLLBACK")


1			"""Handles all postgresql data and tables"""
2			import os
3			import psycopg2
4			import psycopg2.errors
5			from psycopg2.extensions import AsIs
6			from .tables import tables
7			from .logger import make_logger
8
9			# Imports the main logger
10			log = make_logger("database", os.getenv("LOG_LEVEL", "INFO"))
11
12			# Creates the connection to the database
13			connection = psycopg2.connect(os.getenv("DATABASE_URL"), sslmode="require")
14			cursor = connection.cursor()
15
16			DuplicateError = psycopg2.errors.lookup("23505")
17
18
19			def table_create() -> None:
20			"""Table_create
21
22			Creates tables if they do not exist at startup. All tables are pulled from tables.py
23
24			"""
25			for table in tables:
26			cursor.execute(table)
27
28
29			def _format_step(table: str) -> str:
30			"""Format_step
31			---
32
33			Returns the format string to be used in insert. This was split from insert to make it less
34			complex and easier to read.
35
36			Arguments:
37			---
38			table {str} -- The name of the table for the insert
39
40			Returns:
41			---
42			str -- Returns a string that will be used for cursor execution
43			"""
44			if table == "schools":
45			query_str = (
46			"INSERT INTO schools"
47			"(school, region, color, id, added_by, added_by_id) "
48			"VALUES (%s, %s, %s, %s, %s, %s);"
49			)
50			elif table == "errors":
51			query_str = (
52			"INSERT INTO errors"
53			"(id, command, message, error, time, ack) "
54			"VALUES (%s, %s, %s, %s, %s, %s);"
55			)
56			elif table == "reports":
57			query_str = (
58			"INSERT INTO reports"
59			"(id, name, name_id, message, time) "
60			"VALUES (%s, %s, %s, %s, %s);"
61			)
62			elif table == "admin_channels":
63			query_str = (
64			"INSERT INTO admin_channels (name, id, log) "
65			"VALUES (%s, %s, %s) ON CONFLICT DO NOTHING;"
66			)
67			elif table == "bot_admins":
68			query_str = "INSERT INTO bot_admins (name, id) VALUES (%s, %s) ON CONFLICT DO NOTHING;"
69			elif table == "regions":
70			query_str = "INSERT INTO regions (name, id) VALUES (%s, %s)"
71			else:
72			log.error("Table {} not found.".format(table))
73			return "error"
74			return query_str
75
76
77			def _result_parser(column: str, fetched: list) -> list:
78			"""Result_parser
79			---
80
81			Arguments:
82			---
83			column {str} -- The name of the column(s) only used to determine how to parse the results.
84			fetched {list} -- The results pulled from the table.
85
86			Returns:
87			---
88			list -- A normal list or a list of tuples.
89			"""
90			# fetched is a list so it does not need breaking up
91			if column == "*" or column.find(",") != -1:
92			result = fetched
93			# Breaks up the tuples to a standard list.
94			else:
95			result = [x[0] for x in fetched]
96			return result
97
98
99			async def insert(table: str, data: list) -> [None, str]:
100			"""Insert
101			---
102
103			Asynchronous Function
104
105			Inserts a new row to an existing table. Get the string to execute with from :ref:`_format_step`.
106
107			Arguments:
108			---
109			table {str} -- c to perform the insert on
110			data {list} -- The data that gets placed into the format_str
111
112			Returns:
113			---
114			str -- In the event of an error inserting into the table the string 'error' will be
115			returned. If there is no error then 'None' will be returned.
116			PostgreSQL Equivalent:
117			---
118			INSERT into :ref:`table` VALUE (:ref:`*data`);
119			"""
120			format_str = _format_step(table)
121			if format_str == "error":
122			return "error"
123			log.debug("String: {} Data {}".format(format_str, " ".join(map(str, data))))
124			try:
125			# Tables with 6 values
126			if table in ["schools", "errors"]:
127			cursor.execute(format_str, (data[0], data[1], data[2], data[3], data[4], data[5]))
128			# Tables with 5 values
129			elif table in ["reports"]:
130			cursor.execute(format_str, (data[0], data[1], data[2], data[3], data[4]))
131			# Tables with 3 values
132			elif table == "admin_channels":
133			cursor.execute(format_str, (data[0], data[1], data[2]))
134			# Tables with 2 values
135			elif table in ["bot_admins", "regions"]:
136			cursor.execute(format_str, (data[0], data[1]))
137			connection.commit()
138			return None
139			except psycopg2.Error as pge:
140			log.error(pge)
141			cursor.rollback()
142			if isinstance(pge, DuplicateError):
143			return "duplicate"
144			return "error"
145
146
147			async def fetch(table: str, column: str) -> list:
148			"""Fetch
149			---
150
151			Asynchronous Function
152
153			Retrieves values from the :ref:`column` from the :ref:`table`.
154
155			Arguments:
156			---
157			table {str} -- Name of the table that data is being fetched from.
158			column {str} -- The column(s) that is being fetched. Multiple columns need to comma
159			separated, if all columns are wanted then use '*'.
160
161			Returns:
162			---
163			list -- A list of tuples for '*' or multiple columns. For one column it is a list.
164
165			Postgresql Equivalent:
166			---
167			SELECT :ref:`column` from :ref:`table`;
168			"""
169			try:
170			format_str = "SELECT %s FROM %s;"
171			cursor.execute(format_str, (AsIs(column), AsIs(table)))
172			fetched = cursor.fetchall()
173			return _result_parser(column, fetched)
174			except psycopg2.Error as pge:
175			log.error(pge)
176
177
178			async def select(
179			table: str,
180			column: str,
181			where_column: str,
182			where_value: [str, bool, int],
183			symbol: [str, bool] = "=",
184			) -> list:
185			"""Select
186			---
187
188			Asynchronous Function
189
190			Selects one row from the table based on selector.
191
192
193			Arguments:
194			---
195			table {str} -- Name of the table that data is being fetched from.
196			column {str} -- The column(s) that is being fetched. Multiple columns need to comma
197			separated, if all columns are wanted then use '*'.
198			where_column {str} -- The column that is going have the value of :ref:`where_value`.
199			where_value {str} -- The value that you are matching.
200			symbol {str} -- [description] (default: {"="})
201
202			Returns:
203			---
204			list -- List of values that are the results.
205
206			Postgresql Equivalent:
207			---
208			SELECT :ref:`column` FROM :ref:`table` WHERE :ref:`where_column` :ref:`symbol` :ref:`where_value`; # noqa: E501 pylint: disable=line-too-long
209			"""
210			try:
211			format_str = "SELECT %s FROM %s WHERE %s %s %s;"
212			cursor.execute(
213			format_str,
214			(AsIs(column), AsIs(table), AsIs(where_column), AsIs(symbol), where_value),
215			)
216			fetched = cursor.fetchall()
217			return _result_parser(column, fetched)
218			except psycopg2.Error as pge:
219			log.error(pge)
220			cursor.execute("ROLLBACK")
221
222
223			async def update(
224			table: str, column: str, where_value: str, new_value: [str, bool], where_column: str = ""
225			) -> None:
226			"""Update
227			---
228
229			Asynchronous Function
230
231			Updates a value in the table
232
233			Arguments:
234			table {str} -- Name of the table that the data is being updated on.
235			column {str} -- The column that is being updated. Multiple columns are not supported.
236			where_value {str, bool} -- The value that is going to be updated.
237			new_value {str} -- The new value for :ref:`where_value`
238
239			Postgresql Equivalent:
240			---
241			UPDATE :ref:`table` SET :ref:`column` = :ref:`new_value`
242			WHERE :ref:`check_column` = :ref:`where_value`;
243			"""
244			if not where_column:
245			where_column = column
246			try:
247			format_str = "UPDATE %s SET %s = %s where %s = %s"
248			cursor.execute(
249			format_str,
250			(AsIs(table), AsIs(column), new_value, AsIs(where_column), where_value),
251			)
252			connection.commit()
253			except psycopg2.Error as pge:
254			log.error(pge)
255			cursor.execute("ROLLBACK")
256
257
258			async def delete(table: str, column: str, value: str) -> None:
259			"""Delete
260			---
261			Asynchronous Function
262
263			Removes an entry from the table where :ref:`value` is equal.
264
265			Arguments:
266			---
267			table {str} -- Name of the table that the data is being deleted from.
268
269			column {str} -- The column to which the :ref:`value` is going to match.
270
271			value {str} -- The value in the row which is going to match to a value in :ref:`column`
272
273			Postgresql Equivalent:
274			---
275			DELETE FROM :ref:`table` WHERE :ref:`column` = :ref:`value`;
276			"""
277			try:
278			log.info("Deleting {} where {} from {}".format(column, value, table))
279			format_str = "DELETE FROM %s WHERE %s = %s"
280			cursor.execute(format_str, (AsIs(table), AsIs(column), value))
281			connection.commit()
282			except psycopg2.Error as pge:
283			log.error(pge)
284			cursor.execute("ROLLBACK")
285

Competitive-Cyber-Clubs / Discord-Bot

Pull Request — master (#50)

bot.utils.datahandler._format_step() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like