{ "cells": [ { "cell_type": "code", "execution_count": 1, "metadata": {}, "outputs": [], "source": [ "import pymysql.cursors\n", "import sys\n", "\n", "\n", "def get_sql_time(datetime_object):\n", " return datetime_object.strftime('%Y-%m-%d %H:%M:%S')\n", "\n", "\n", "class SQLInjectionError(Exception):\n", " def __init__(self):\n", "\n", " # Call the base class constructor with the parameters it needs\n", " super().__init__(\"Detected possible SQL injection attack!\")\n", "\n", "\n", "class DatabaseConnection(object):\n", " \"\"\"\n", " a singleton class for a global database connection\n", " \"\"\"\n", "\n", " instance = None\n", "\n", " @staticmethod\n", " def global_cursor():\n", " assert DatabaseConnection.instance is not None\n", " return DatabaseConnection.instance.get_cursor()\n", "\n", " @staticmethod\n", " def global_close():\n", " assert DatabaseConnection.instance is not None\n", " DatabaseConnection.instance.close()\n", "\n", " @staticmethod\n", " def global_commit():\n", " assert DatabaseConnection.instance is not None\n", " DatabaseConnection.instance.commit()\n", "\n", " @staticmethod\n", " def global_ping():\n", " assert DatabaseConnection.instance is not None\n", " DatabaseConnection.instance.connection.ping()\n", "\n", " @staticmethod\n", " def global_single_query(query, params=None):\n", " DatabaseConnection.global_ping()\n", " if ';' in query:\n", " # Possible injection!\n", " raise SQLInjectionError()\n", "\n", " with DatabaseConnection.global_cursor() as c:\n", " if params is None:\n", " c.execute(query)\n", " else:\n", " c.execute(query, params)\n", "\n", " return c.fetchall()\n", "\n", " @staticmethod\n", " def global_single_execution(sql_statement, params=None):\n", " DatabaseConnection.global_ping()\n", " if ';' in sql_statement:\n", " # Possible injection detected!\n", " raise SQLInjectionError()\n", "\n", " with DatabaseConnection.global_cursor() as c:\n", " if params is None:\n", " c.execute(sql_statement)\n", " else:\n", " c.execute(sql_statement, params)\n", " DatabaseConnection.global_commit()\n", "\n", " def __init__(self,\n", " host: str,\n", " port: int,\n", " user: str,\n", " password: str,\n", " db: str,\n", " charset: str):\n", "\n", " assert DatabaseConnection.instance is None\n", " try:\n", " self.connection = pymysql.connect(\n", " host=host,\n", " port=port,\n", " user=user,\n", " password=password,\n", " db=db,\n", " charset=charset,\n", " cursorclass=pymysql.cursors.DictCursor)\n", " DatabaseConnection.instance = self\n", " except Exception as e:\n", " sys.stderr.write(\"could not connect to database '\" +\n", " str(db) +\n", " \"' at \" +\n", " user +\n", " \"@\" +\n", " host +\n", " \":\" +\n", " str(port) +\n", " \"\\nCheck the configuration in settings.py!\\n\")\n", " raise Exception('could not connect to database')\n", "\n", " def get_cursor(self):\n", " return self.connection.cursor()\n", "\n", " def close(self):\n", " self.connection.close()\n", " DatabaseConnection.instance = None\n", "\n", " def commit(self):\n", " self.connection.commit()\n", "\n", "\n", "def test_connection():\n", " import db_settings as settings\n", " DatabaseConnection(settings.db_host,\n", " settings.db_port,\n", " settings.db_user,\n", " settings.db_pw,\n", " settings.db_db,\n", " settings.db_charset)" ] }, { "cell_type": "code", "execution_count": 2, "metadata": {}, "outputs": [], "source": [ "test_connection()" ] }, { "cell_type": "code", "execution_count": 3, "metadata": {}, "outputs": [ { "data": { "text/plain": [ "()" ] }, "execution_count": 3, "metadata": {}, "output_type": "execute_result" } ], "source": [ "DatabaseConnection.global_single_query(\"show tables\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.7.3" } }, "nbformat": 4, "nbformat_minor": 4 }