I have a repository project, which is taking me longer than I expected to setup both database and the task (a cron job) both in a smooth way. To test connection, first run the postgres server then create the database:

docker run --rm -d --name postgres -e POSTGRES_USER=myuser -e POSTGRES_PASSWORD=mypassword -p 5432:5432 postgres
docker exec -it postgres psql -U myuser -c 'CREATE DATABASE mydb;'

This step is somewhat already manual. For that, I wrote a docker-compose, but still stumble on errors like ‘Cannot create a database within a transaction’. Anyhow, after run of commands above, we will (or should) be able to run to run a simple script like:

from sqlalchemy import create_engine
from sqlalchemy.sql import text

engine = create_engine("postgresql+psycopg2://myuser:[email protected]:5432/mydb")

def app():
    with engine.connect() as conn:
        stmt = text("select * from pg_database")
        print(conn.execute(stmt).fetchall())


if __name__ == "__main__":
    app()

My initial idea was to parametrize the test within a conftest.py. However, when I import the model from relative path src.database.models in another test file (or in this case conftest), I am not able to because of error log ModuleNotFoundError: No module named 'src'.

from sqlalchemy import create_engine
from sqlalchemy.sql import select
from sqlalchemy.orm import sessionmaker

from src.database.models import UserDB
from datetime import datetime

engine = create_engine("postgresql+psycopg2://myuser:[email protected]:5432/mydb")

Session = sessionmaker(bind=engine)


def app():
    with Session() as session:
        session.add_all(
            [
                UserDB(
                    id=1,
                    username="John Smith",
                    email="[email protected]",
                    password="mypwd1234!",
                    created_at=datetime.now(),
                    updated_at=datetime.now()
                ),
                UserDB(
                    id=2,
                    username="Amy Johnson",
                    email="[email protected]",
                    password="mypwd1234?",
                    created_at=datetime.now(),
                    updated_at=datetime.now()
                ),
            ]
        )
        session.commit()



    with engine.connect() as conn:
        stmt = select(UserDB)
        print(conn.execute(stmt).fetchall())


if __name__ == "__main__":
    app()

It is indeed embarrasing to stumble upon such trivial errors, even after searching for 2 days in a row. I can cite other errors as well, but if you can help me with this, it will propel me in further investigations.

[1] https://github.com/brunolnetto/db-cron-docker