master-thesis/db/database_connection.ipynb

191 lines
5.7 KiB
Plaintext

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