PyMySQL
https://github.com/PyMySQL/PyMySQL
https://pymysql.readthedocs.io/en/latest
https://pypi.org/project/PyMySQL
pip install PyMySQL==0.9.3
Connections Object
https://pymysql.readthedocs.io/en/latest/modules/connections.html
Cursors Object
https://pymysql.readthedocs.io/en/latest/modules/cursors.html
Getting Started
import pymysql.cursors
# Connect to the database
connection = pymysql.connect(host='localhost',
user='user',
password='passwd',
db='db',
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor)
try:
with connection.cursor() as cursor:
# Create a new record
sql = "INSERT INTO `users` (`email`, `password`) VALUES (%s, %s)"
cursor.execute(sql, ('webmaster@python.org', 'very-secret'))
# connection is not autocommit by default. So you must commit to save your changes.
connection.commit()
with connection.cursor() as cursor:
# Read a single record
sql = "SELECT `id`, `password` FROM `users` WHERE `email`=%s"
cursor.execute(sql, ('webmaster@python.org',))
result = cursor.fetchone()
print(result)
finally:
connection.close()
Connection Pooling
https://pythontic.com/database/mysql/connection%20pooling
Singleton Class
import pymysql.cursors
import logging
from constants import (
AURORA_DB_DBNAME,
AURORA_DB_HOST,
AURORA_DB_PASSWORD,
AURORA_DB_USER,
AWS_ACCESS_KEY_ID,
AWS_SECRET_ACCESS_KEY,
DEBUG,
)
mysql_conn = None
def getdb():
global mysql_conn
if not mysql_conn:
try:
mysql_conn = pymysql.connect(
host=AURORA_DB_HOST,
user=AURORA_DB_USER,
password=AURORA_DB_PASSWORD,
db=AURORA_DB_DBNAME,
charset="utf8mb4",
cursorclass=pymysql.cursors.DictCursor,
)
except Exception as e:
logging.exception(f"Some error in establishing mysql connection.")
raise e
return mysql_conn
DBClass
import sys
import pymysql
import logging
class Database:
def __init__(self, config):
self.host = config.db_host
self.username = config.db_user
self.password = config.db_password
self.port = config.db_port
self.dbname = config.db_name
self.conn = None
def open_connection(self):
"""Connect to MySQL Database."""
try:
if self.conn is None:
self.conn = pymysql.connect(self.host,
user=self.username,
passwd=self.password,
db=self.dbname,
connect_timeout=5)
except pymysql.MySQLError as e:
logging.error(e)
sys.exit()
finally:
logging.info('Connection opened successfully.')
def run_query(self, query):
"""Execute SQL query."""
try:
self.open_connection()
with self.conn.cursor() as cur:
if 'SELECT' in query:
records = []
cur.execute(query)
result = cur.fetchall()
for row in result:
records.append(row)
cur.close()
return records
else:
result = cur.execute(query)
self.conn.commit()
affected = f"{cur.rowcount} rows affected."
cur.close()
return affected
except pymysql.MySQLError as e:
print(e)
finally:
if self.conn:
self.conn.close()
self.conn = None
logging.info('Database connection closed.')
class PostgresClient:
host: str
database_name: str
table_name: str
port: int
username: str
password: str
conn = None
def __init__(self, host, database_name, port, username,
password):
self.host = host
self.database_name = database_name
self.port = port
self.username = username
self.password = password
def connect(self):
try:
logger.info(f"connecting to {self.host}")
self.conn = pg8000.connect(
host=self.host, port=self.port, database=self.database_name,
user=self.username, password=self.password
)
return True
except Exception as e:
logger.error(f"Couldn't connect to the db {self.host}- {e}")
return False
def insert(self, insert_strings: list, record_to_insert: tuple):
if not self.conn:
logger.error("No able to connect to the database")
return None
cur = self.conn.cursor()
for insert_string in insert_strings:
cur.execute(f"{insert_string}", record_to_insert)
self.conn.commit()
cur.close()
def close(self):
if self.conn:
self.conn.close()
self.conn = None
https://hackersandslackers.com/python-mysql-pymysql
https://medium.com/@vipinc.007/python-a-database-interaction-class-using-pymysql-3338fb90f38c
Others
https://pypi.org/project/mysql-connector-python
GitHub - long2ice/asyncmy: A fast asyncio MySQL/MariaDB driver with replication protocol support
psycopg2
Psycopg is the most popular PostgreSQL database adapter for the Python programming language. Its main features are the complete implementation of the Python DB API 2.0 specification and the thread safety (several threads can share the same connection). It was designed for heavily multi-threaded applications that create and destroy lots of cursors and make a large number of concurrent "INSERT"s or "UPDATE"s.
Psycopg 2 is mostly implemented in C as a libpq wrapper, resulting in being both efficient and secure. It features client-side and server-side cursors, asynchronous communication and notifications, "COPY TO/COPY FROM" support. Many Python types are supported out-of-the-box and adapted to matching PostgreSQL data types; adaptation can be extended and customized thanks to a flexible objects adaptation system.
Psycopg 2 is both Unicode and Python 3 friendly.
https://pypi.org/project/psycopg2-binary