Created
April 1, 2020 14:37
-
-
Save ozturkoktay/706ce877fb693c3b2770f964d40236a0 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!usr/bin/env python | |
# -*- coding: utf8 -*- | |
#: Imports | |
import MySQLdb | |
# class: Connection | |
# A simple mysql connection | |
# @completed | |
class Connection: | |
# method: __init__ | |
# @completed | |
def __init__(self): | |
#: Create the connection | |
self.DBCONNECTION = MySQLdb.connect( | |
user='usernama', password='password', | |
host='localhost', database=''db, | |
charset='utf8mb4', use_unicode=True | |
) | |
#: Create the connection cursor | |
self.MYCURSOR = self.DBCONNECTION.cursor() | |
# method: __del__ | |
# @completed | |
def __del__(self): | |
self.DBCONNECTION.close() | |
# method: dbExecute | |
# Executes the sql | |
# @sql, str: The sql statement | |
# @params, tuple: The arguments | |
# @completed | |
def dbExecute(self, sql: str, params: tuple): | |
#: Exec | |
self.MYCURSOR.execute(sql, params) | |
# method: dbInsert | |
# Inserts an item and returns its id | |
# @sql, str: The sql statement | |
# @params, tuple: The arguments | |
# @return, int: The id of the row | |
# @completed | |
def dbInsert(self, sql: str, params: tuple) -> int: | |
#: Exec | |
self.dbExecute(sql, params) | |
#: Return last row id | |
return self.MYCURSOR.lastrowid | |
# method: dbSelect | |
# Selects an sql query | |
# @sql, str: The sql statement | |
# @params, tuple: The arguments | |
# @return, sqlobject: The result | |
# @completed | |
def dbSelect(self, sql: str, params: tuple): | |
#: Exec | |
self.dbExecute(sql, params) | |
#: Fetch the results | |
myresult = self.MYCURSOR.fetchall() | |
#: Return them | |
return myresult | |
# method: dbList | |
# Loads a list from dataset | |
# @sql, str: The input sql string | |
# @params, tuple: The parameters | |
# @return, list: The output list | |
# @completed | |
def dbList(self, sql: str, params: tuple = None) -> list: | |
#: Get the rows | |
rows = self.dbSelect(sql, params) | |
#: Create list | |
lst = [] | |
#: Loop for each | |
for r in rows: lst.append(r[0]) | |
#: Return the output | |
return lst | |
# method: dbDict | |
# Loads a dict from dataset | |
# @sql, str: The input sql string | |
# @params, tuple: The parameters | |
# @return, dict: The output list | |
# @completed | |
def dbDict(self, sql: str, params: tuple = None) -> dict: | |
#: Get the rows | |
rows = self.dbSelect(sql, params) | |
#: Create list | |
lst = {} | |
#: Loop for each | |
for r in rows: | |
lst[r[0]] = r[1] | |
#: Return the output | |
return lst | |
Connection() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment