Die Datenbank kann bei Verwendung sqlalchemy nicht inspiziert werden ⇐ Python

[Anonymous]

Ich habe eine Postgres -Datenbank in einem Docker -Container. Ich komponiere es mit diesem Docker-compose.yaml: < /p>

Code: Select all

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:

< /code>
Ich erhalte die Umgebungsvariablen aus einer .Env -Datei. Dann habe ich ein Python -Skript
, das SQLAlchemy verwendet, um ein Tabellenschema zu erstellen, und eine Funktion, um die Tabelle wie folgt zu initieren: < /p>
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"""

"""

@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.")
< /code>
Dann habe ich die folgende Funktion, um Daten an die Datenbank zu senden < /p>
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()

Ich erhalte diesen Fehler jedoch, wenn der Befehl calldata.init_table () ausgeführt wird. Partikel, wenn nicht inspizieren (Motor) .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)

< /code>
Ich habe wieder nach unten und nach oben komponiert und habe den gleichen Fehler. Ich denke, das liegt daran, dass ich das Passwort bei der Überprüfung irgendwie analysieren muss. Wie kann ich es lösen?