Conexão Banco MySql (Class)
Bibliotecas
def insert(self, df, tabela):
df.to_sql(tabela, if_exists = 'append', index = False, con = self.con)
def exc_query(self, query):
aa = self.con.execute(query)
return aa
def delete(self, query):
self.con.execute(query)
def __del__(self):
try:
self.con.close()
except:
No
import pandas as pd
import urllib.parse
import datetime as dt
from sqlalchemy import create_engine
import urllib.parse
import datetime as dt
from sqlalchemy import create_engine
Classe
class ConnBanco:
def __init__(self, user, password, host, port, database, dialects):
self.user = user
self.password = urllib.parse.quote_plus(password)
self.host = host
self.port = port
self.database = database
self.dialects = dialects
self.engine = create_engine(f"{self.dialects}://{self.user}:{self.password}@{self.host}:{self.port}/{self.database}") #, echo=True
def __init__(self, user, password, host, port, database, dialects):
self.user = user
self.password = urllib.parse.quote_plus(password)
self.host = host
self.port = port
self.database = database
self.dialects = dialects
self.engine = create_engine(f"{self.dialects}://{self.user}:{self.password}@{self.host}:{self.port}/{self.database}") #, echo=True
self.con = self.engine.connect()
def select(self, query):
Result = pd.read_sql(query, con = self.con)
return Result
def insert(self, df, tabela):
df.to_sql(tabela, if_exists = 'append', index = False, con = self.con)
def exc_query(self, query):
aa = self.con.execute(query)
return aa
def delete(self, query):
self.con.execute(query)
def __del__(self):
try:
self.con.close()
except:
No
Conexão
con = ConnBanco('usuario', 'Senha', '127.0.0.1', 3306, 'DataBase', 'mysql+pymysql')
query = 'Select * FROM Tabela;'
retorno = con.select(query)
retorno
Comentários
Postar um comentário