import pymysql #import gn_mysql import decimal import configparser # Usei como base esse link # https://cafeinacodificada.com.br/python-mysql/ import datetime # class gn_mysql(): """Classe generica para uso do banco de dados mysql""" def __init__(self, cliente:str="", db:str="", debug:bool=False): # self.resetvadWorlds = ('now()') if (cliente) : self.cliente = cliente self.db = db self.debug = debug #pprint(self.conexao) self.connectByCliente(cliente, db) def connectByCliente(self, cliente:str, db:str): path = f'/user/bindib/aplicativos/{cliente.lower()}/appConfig.ini'#.format(cliente.lower()) #if self.debug : # path = '\\\\192.168.1.3\\raiz\\user\\bindib\\aplicativos\\{0}\\appConfig.ini'.format(cliente.lower()) with open(path) as _config: dados = _config.read() #getConfig = lambda search, default=False: dados.split('{}='.format(search))[1].split('\n')[0] or default def get_config(key, **kwargs) : split_key = dados.split(f'{key}=') try: rest = split_key[1] return kwargs.get("type", str)(rest.split('\n')[0]) except : return kwargs.get("default",'') #william usuario = get_config('usuario') senha = get_config('senha') host = get_config('host' , default="localhost") port = get_config('port' , default=3306, type=int) db = get_config('db' , default=db) ssl_ca = get_config('ssl_ca' , default='') ssl_cert = get_config('ssl_cert' , default='') ssl_key = get_config('ssl_key' , default='') ssl_verify_cert = get_config('ssl_verify_cert' , default=False) """ if self.debug and host == 'localhost' : import os from dotenv import load_dotenv load_dotenv() usuario = os.getenv("USUARIO") senha = os.getenv("SENHA") host = os.getenv("HOST") """ #GRANT ALL ON *.* TO 'api'@'%' IDENTIFIED BY 'OGF1lCkVJyx4OoHd' WITH GRANT OPTION; self.pydb_connection(host=host,user=usuario,passwd=senha,db=db,port=port,ssl_ca=ssl_ca,ssl_cert=ssl_cert,ssl_key=ssl_key,ssl_verify_cert=ssl_verify_cert) def pydb_connection(self, **kwargs): self.conexao = pymysql.connect( host = kwargs.get("host" ) , port = kwargs.get("port" ,3306) , user = kwargs.get("user" ) , passwd = kwargs.get("passwd" ) , db = kwargs.get("db" ) , autocommit = kwargs.get("autocommit" ,True) , ssl = { "ca" : kwargs.get("ssl_ca" ,None) , "cert" : kwargs.get("ssl_cert" ,None) , "key" : kwargs.get("ssl_key" ,None) , "verify_cert" : kwargs.get("ssl_verify_cert",False) , "check_hostname" : False } if kwargs.get("ssl_ca" ,None) else None ) def select(self, sql:str, params:tuple = ()) -> list: """ Metodo para select @param {String} sql - sql a ser executado @return {Array} retorno - array com os dados do retorno do sql """ # inicia a variavel resultado resultado = False #return # Cria um cursor: # pymysql.cursors.DictCursor para que ele retorne um json chave:valor result, error = self.executesql(sql, params) if not result : return error return result.fetchall() """ with self.conexao.cursor(pymysql.cursors.DictCursor) as cursor: # Executa o comando: try: cursor.execute(sql) # Recupera o resultado: resultado = cursor.fetchall() except pymysql.InternalError as InternalError : print(str(InternalError)) return """ #corrige os dados com tipos que viriam a dar problema na conversao json #for row in resultado: # Corrije os pontos com problemas # row = self.fixTypes(row) #retorna o resultado da consulta def simple_select(self, **kwargs) : where=" AND ".join(kwargs.get('where',[])) where_query = f"{where}"##"" if "where" not in kwargs else "WHERE " + " AND ".join(kwargs['where']) fields = ",".join(kwargs.get("fields", ['*']))##"*" if "fields" not in kwargs else ",".join(kwargs['fields']) table = kwargs.get('table','')##if "table" not in kwargs else kwargs['table'] where = (where_query and "WHERE " + where_query) or '' sql = f"SELECT {fields} FROM {table} {where}" result, error = self.executesql(sql) if error : return error if "onlyFirst" in kwargs and kwargs['onlyFirst'] == True : return result.fetchone() or {} else : return result.fetchall() def delete(self, table, where=["1=1"]): #with self.conexao.cursor() as cursor: # Executa o comando: query_and = " AND ".join(where) sql = f"DELETE FROM {table} WHERE {' AND '.join(where)}" return self.executesql(sql) def update(self, table:str, data:list, where = []): """Receme um objeto ou array de objetos e monta o insert no banco Arguments: table {str} -- Tabela que serao inseridos os dados data {[{}]} -- objeto com os dados, tambem aceita um array de objetos """ if isinstance(data, list): pass updates =[] #insert=[] #values=[] for col in data: val = data[col] val = self.conexao.escape_string(str(val)) if val not in self.resetvadWorlds: val = f"'{val}'"#.format(val) #insert.append(col) #values.append(val) updates.append(f"{col}={val}")#.format(col, val)) #print(','.join(updates)) sql = f''' UPDATE {table} SET {','.join(updates)} WHERE 1=1 AND {" AND ".join(where)} '''#.format(table, , ) return self.executesql(sql) def insert(self, table:str, data:list): """Receme um objeto ou array de objetos e monta o insert no banco Arguments: table {str} -- Tabela que serao inseridos os dados data {[{}]} -- objeto com os dados, tambem aceita um array de objetos """ if isinstance(data, list): pass insert=[] values=[] for col in data: val = data[col] val = self.conexao.escape_string(str(val)) if val not in self.resetvadWorlds: val = f"'{val}'"#.format(val) insert.append(col) values.append(val) sql = f''' INSERT INTO {table} ( {','.join(insert)} ) VALUES( {','.join(values)} ) '''#.format(table, , ) # conexao return self.executesql(sql) def callProc (self, sql:str, params:tuple = ()): args = [params] with self.conexao.cursor(pymysql.cursors.DictCursor) as cursor: cursor.callproc(sql, args) return cursor.fetchall() #corrige os dados com tipos que viriam a dar problema na conversao json #for row in resultado: # Corrije os pontos com problemas #row = self.fixTypes(row) #retorna o resultado da consulta #return resultado def executesql(self, sql:str, params:tuple = () ) : # conexao with self.conexao.cursor(pymysql.cursors.DictCursor) as cursor: try: # Executa o comando: if not params : cursor.execute(sql) else: cursor.execute(sql, params) except pymysql.Error as MySQLError : return False, MySQLError #exit() return cursor, False pass def insertList(self, table:str, arr:list) -> None: ''' INSERE UMA LISTA TODO: melhorias de insert hoje faz de um a um o insert, implementar para fazer todos de uma vez, usar parametro no insert para dizer se deve ou nao inserir ''' for insert in arr: self.insert(insert, insert) #!Deprecated def fixTypes(self, row) -> object : for key in row: if isinstance(row[key], decimal.Decimal) : row[key] = float(row[key]) #elif row[key] == None: # row[key] = "" elif isinstance(row[key], datetime.date) : row[key] = str(row[key]) key, row[key] = self.fixTypes_customization(key, row[key]) return row def fixTypes_customization(self, name, value) : return name, value def __del__(self): """Destrutor fecha a conexao com o banco""" # Finaliza a conexão self.conexao.close() if __name__ == '__main__': from pprint import pprint # # banco = gn_mysql(db='ebs_app_pedidos3', user='root', passwd='cedros') banco = gn_mysql('JVMALDANER', 'ebs_app_pedidos3') # resultado = banco.select("SELECT * FROM TBCONDPGTO") # pprint(resultado)