master-thesis/db/database_connection.ipynb

{
 "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
}