I have a postgres database inside a docker container. I compose it up with this docker-compose.yaml:
services:
db:
container_name: postgres
image: postgres:latest
restart: always
volumes:
- db-data:/var/lib/postgresql/data
environment:
POSTGRES_DB: ${POSTGRES_DB}
POSTGRES_USER: ${POSTGRES_USER}
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
POSTGRES_HOST: ${POSTGRES_HOST}
POSTGRES_PORT: ${POSTGRES_PORT}
healthcheck:
test: [ "CMD", "pg_isready" , "-d", "${POSTGRES_DB}", "-U", "${POSTGRES_USER}"]
# The `interval` option specifies how often to run the health check.
interval: 10s
timeout: 5s
retries: 5
volumes:
db-data:
I get the environment variables from an .env file. Then I have a python script that uses SQLAlchemy to create table schema and a function to init the table like this:
from sqlalchemy import Column, Integer, String, DateTime, Float
from sqlalchemy import inspect
from sqlalchemy.orm import Mapped, DeclarativeBase
from pgvector.sqlalchemy import Vector
class CallData(DeclarativeBase):
__tablename__ = "call_data"
id: Mapped[int] = Column(Integer, primary_key=True)
nombre: Mapped[str] = Column(String(255), nullable=False)
entidad: Mapped[str] = Column(String(255), nullable=False)
descripcion = Vector(dim=300, nullable=False)
def __repr__(self):
return f"""
<CallData(id={self.id}, titulo={self.nombre}, entidad_convocante={self.entidad},
descripcion={self.descripcion})>
"""
@classmethod
def init_table(cls, engine):
"""Initialize the table in the database."""
if not inspect(engine).has_table(cls.__tablename__):
cls.metadata.create_all(engine)
print(f"Table {cls.__tablename__} created.")
else:
print(f"Table {cls.__tablename__} already exists.")
Then I have the following fucntion to send data to the database
def send_to_db(contenido):
load_dotenv()
engine = create_engine(
f"postgresql+psycopg://{os.getenv("POSTGRES_USER")}:{os.getenv("POSTGRES_PASSWORD")}@{os.getenv("POSTGRES_HOST)}:{os.getenv("POSTGRES_PORT")}/{os.getenv("POSTGRES_DB")}",
)
Session = sessionmaker(bind=engine)
CallData.init_table(engine)
with Session() as session:
# Assuming `contenido` is a list of dictionaries with the data to be inserted
for entry in contenido:
call_data = CallData(
nombre=entry["convocatoria"],
entidad=entry["entidad"],
descripcion=entry["descripcion"],
)
session.add(call_data)
session.commit()
However, I am getting this error when the CallData.init_table()
command is run. Particulary, on if not inspect(engine).has_table(self.__tablename__)
(psycopg.OperationalError) connection failed: connection to server at ${POSTGRES_HOST}, port ${POSTGRES_PORT} failed: FATAL: password authentication failed for user ${POSTGRES_USER}
connection to server at ${POSTGRES_HOST}, port ${POSTGRES_PORT} failed: FATAL: password authentication failed for user ${POSTGRES_USER}"
(Background on this error at: https://sqlalche.me/e/20/e3q8)
I have composed down and up again, and I got the same error. I think it's because I have to parse through the password on inspect, somehow. How can I solve it?
inspect
, the error is happening there because it is the first time that the engine tries to connect to the database. Trying printing the environment variables before you create the engine and see if they are being populated correctly.os.getenv
are not returning the correct values. Try printing that complete Fstring at the point that you callcreate_engine
.