utils.datahandler._format_step() - Code Metrics - Inspection of "Merge pull request #36 from Competitive-Cyber-Club..." - Competitive-Cyber-Clubs/Discord-Bot - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( e4d30e...a5a665 )

unknown

created 2021-04-12 23:56 UTC

utils.datahandler._format_step() B

↳ Parent: 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
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()


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

Competitive-Cyber-Clubs / Discord-Bot

Push — master ( e4d30e...a5a665 )

utils.datahandler._format_step() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like