Aprenda a conectarse a una base de datos PostgreSQL usando Python y SQLAlchemy. SQLAlchemy nos permite abstraer gran parte de la implementación para que podamos trabajar con Python para administrar los objetos de nuestra base de datos.
Puedes descargar el código en el Analizando Alpha GitHub Repo.
Primero, importe las importaciones necesarias para que podamos crear un motor de base de datos y una sesión desde ese motor.
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy_utils import database_exists, create_database
from local_settings import postgresql as settingsNotará que local_settings contiene la configuración de PostgreSQL. Es solo un archivo de diccionario y puedes ver un ejemplo en el repositorio de GitHub.
Crea una función get_engine para hacernos la vida más fácil.
def get_engine(user, passwd, host, port, db):
url = f"postgresql://{user}:{passwd}@{host}:{port}/{db}"
if not database_exists(url):
create_database(url)
engine = create_engine(url, pool_size=50, echo=False)
return engineObtenga las credenciales de la importación local_settings anterior.
engine = get_engine(settings['pguser'],
settings['pgpasswd'],
settings['pghost'],
settings['pgport'],
settings['pgdb'])Verifique que el motor se haya creado correctamente.
Cree una nueva función para que no tengamos que ingresar nuestras credenciales cada vez.
def get_engine_from_settings():
keys = ['pguser','pgpasswd','pghost','pgport','pgdb']
if not all(key in keys for key in settings.keys()):
raise Exception('Bad config file')
return get_engine(settings['pguser'],
settings['pgpasswd'],
settings['pghost'],
settings['pgport'],
settings['pgdb'])Cree una función de sesión que se vincule a nuestro motor creado anteriormente.
def get_session():
engine = get_engine_from_settings()
session = sessionmaker(bind=engine)()
return sessionCree una nueva sesión y verifique que funcione.
Felicitaciones, se conectó con éxito a Postgres usando Python y SQLALchemy.
Cuando haya terminado, asegúrese de cerrar su sesión.
También puede cerrar todas las sesiones registradas actualmente.
engine = session.get_bind()
engine.dispose() # Close all checked in sessions