Last active
January 26, 2023 16:24
-
-
Save dario61081/d31637b845e9fc18867e9bf0cef835c3 to your computer and use it in GitHub Desktop.
Database class for sqlalchemy database manager
This file contains 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
class Database: | |
def __init__(self, **kwargs): | |
self.username = kwargs.get('username', 'root') | |
self.password = kwargs.get('password', '') | |
self.host = kwargs.get('host', 'localhost') | |
self.port = kwargs.get('port', '3306') | |
self.database = kwargs.get('database', '') | |
self.driver = kwargs.get('driver', 'mysql+pymysql') | |
self.uri = f"{self.driver}://{self.username}:{self.password}@{self.host}:{self.port}/{self.database}" | |
self.engine = create_engine(self.uri, poolclass=None) | |
self.cn = self.engine.connect() | |
self.metadata = MetaData(bind=self.engine) | |
# self.metadata.reflect(bind=self.engine) | |
def execute(self, query: str, **kwargs): | |
return self.cn.execute(text(query), **kwargs) | |
def fetch_all(self, query: str, **kwargs): | |
return self.execute(query, **kwargs).fetch_all() | |
def fetch_one(self, query: str, **kwargs): | |
return self.execute(query, **kwargs).fetch_one() | |
def fetch_scalar(self, query: str, **kwargs): | |
return self.execute(query, **kwargs).scalar() | |
def get_session(self): | |
return sessionmaker(self.engine)() | |
def secure_password(self, passwrd: str): | |
return hashlib.sha512(passwrd.encode()).hexdigest() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment