How to use get_connection_string method in testcontainers-python

Best Python code snippet using testcontainers-python_python

config.py

Source:config.py

1import os2from dotenv import load_dotenv3load_dotenv()4def get_connection_string(engine: str, driver: str, user: str, password: str, host: str, port: str, db_name: str) -> str:5    return f"{engine}+{driver}://{user}:{password}@{host}:{port}/{db_name}"6def get_clide_connection_string() -> str:7    return get_connection_string(8        engine="postgresql",9        driver="psycopg2",10        user="postgres",11        password="password",12        host="127.0.0.1",13        port=os.getenv("CLIDE_PORT"),14        db_name=os.getenv("CLIDE_DB_NAME")15    )16def get_climsoft_4_1_1_connection_string() -> str:17    return get_connection_string(18        engine="mysql",19        driver="mysqldb",20        user="root",21        password="password",22        host="127.0.0.1",23        port=os.getenv("CLIMSOFT_4_1_1_PORT"),24        db_name=os.getenv("CLIMSOFT_DB_NAME")25    )26def get_mch_english_connection_string(port_override: str = None) -> str:27    return get_connection_string(28        engine="mysql",29        driver="mysqldb",30        user="root",31        password="password",32        host="127.0.0.1",33        port=os.getenv("MCH_ENGLISH_PORT") if port_override is None else port_override,34        db_name=os.getenv("MCH_DB_NAME")35    )36def get_midas_connection_string(port_override: str = None) -> str:37    return get_connection_string(38        engine="postgresql",39        driver="psycopg2",40        user="postgres",41        password="password",42        host="127.0.0.1",43        port=os.getenv("MIDAS_PORT") if port_override is None else port_override,44        db_name=os.getenv("MIDAS_DB_NAME")45    )46def get_surface_connection_string(port_override: str = None) -> str:47    return get_connection_string(48        engine="postgresql",49        driver="psycopg2",50        user="postgres",51        password="password",52        host="127.0.0.1",53        port=os.getenv("SURFACE_PORT", 45432) if port_override is None else port_override,54        db_name=os.getenv("SURFACE_DB_NAME", "postgres")...

connection_handler.py

Source:connection_handler.py

...3import os4import psycopg25import psycopg2.extras6import configparser7def get_connection_string():8    config = configparser.ConfigParser()9    config.read('db_config')10    user_name = config['POSTGRESQL']['user_name']11    password = config['POSTGRESQL']['password']12    host = config['POSTGRESQL']['host']13    database_name = config['POSTGRESQL']['database_name']14    env_variables_defined = user_name and password and host and database_name15    if env_variables_defined:16        # this string describes all info for psycopg2 to connect to the database17        return 'postgresql://{user_name}:{password}@{host}/{database_name}'.format(18            user_name=user_name,19            password=password,20            host=host,21            database_name=database_name22        )23    else:24        raise KeyError('Some necessary environment variable(s) are not defined')25def open_database():26    try:27        connection_string = get_connection_string()28        connection = psycopg2.connect(connection_string)29        connection.autocommit = True30    except psycopg2.DatabaseError as exception:31        print('Database connection problem')32        raise exception33    return connection34def connection_handler(function):35    def wrapper(*args, **kwargs):36        connection = open_database()37        # we set the cursor_factory parameter to return with a RealDictCursor cursor (cursor which provide dictionaries)38        dict_cur = connection.cursor(cursor_factory=psycopg2.extras.RealDictCursor)39        ret_value = function(dict_cur, *args, **kwargs)40        dict_cur.close()41        connection.close()42        return ret_value43    return wrapper...

db.py

Source:db.py

1from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine2from sqlalchemy.orm import sessionmaker3from sqlalchemy.pool import NullPool, QueuePool4_connection_pool = None5def get_connection_string(read_only: bool = False) -> str:6    """7    Construct the DB URL instead of relying on the string in alembic.ini8    :return: Postgres URL9    """10    return "postgresql+asyncpg://localhost:5432/fastapitest"11# def create_engine() -> AsyncEngine:12#     # pytest-asyncio creates a new event loop per test. Connection pooling with asyncpg does not work with multiple13#     # event loops: https://docs.sqlalchemy.org/en/14/orm/extensions/asyncio.html#using-multiple-asyncio-event-loops14#     # so if the environment is local or test, disable the connection pool.15#     #16#     # QueuePool is the default connection pool and will be used for dev, staging, and production.17#     engine_params = {"future": True, "pool_pre_ping": True, "echo": True, "echo_pool": 'debug'}  # type: Dict[str, Any]18#     if ENV in ("local", "test"):19#         engine_params["poolclass"] = NullPool20#     else:21#         engine_params["poolclass"] = QueuePool22#         engine_params["pool_size"] = db_config["pool"]["min_size"].get(20)23#         engine_params["max_overflow"] = 024#25#     return create_async_engine(get_connection_string(True), **engine_params)26_engine = create_async_engine(get_connection_string(),27                              future=True,28                              pool_pre_ping=True,29                              echo=True,30                              echo_pool='debug',31                              pool_size=5,32                              max_overflow=0)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.