|
| 1 | +import os |
| 2 | +from typing import Optional |
| 3 | +from testcontainers.core.generic import DbContainer |
| 4 | +# from testcontainers.core.utils import raise_for_deprecated_parameter |
| 5 | + |
| 6 | + |
| 7 | +class IRISContainer(DbContainer): |
| 8 | + """ |
| 9 | + InterSystems IRIS database container. |
| 10 | +
|
| 11 | + Example: |
| 12 | +
|
| 13 | + The example spins up a IRIS database and connects to it using the :code:`intersystems-iris` |
| 14 | + driver. |
| 15 | +
|
| 16 | + .. doctest:: |
| 17 | +
|
| 18 | + >>> from testcontainers.iris import IRISContainer |
| 19 | + >>> import sqlalchemy |
| 20 | +
|
| 21 | + >>> iris_container = IRISContainer("intersystemsdc/iris-community:latest") |
| 22 | + >>> with iris_container as iris: |
| 23 | + ... engine = sqlalchemy.create_engine(iris.get_connection_url()) |
| 24 | + ... with engine.begin() as connection: |
| 25 | + ... result = connection.execute(sqlalchemy.text("select $zversion")) |
| 26 | + ... version, = result.fetchone() |
| 27 | + >>> version |
| 28 | + 'IRIS for UNIX (Ubuntu Server LTS for ARM64 Containers) 2023.2 (Build 227U) Mon Jul 31 2023 17:43:25 EDT' |
| 29 | + """ |
| 30 | + def __init__(self, image: str = "intersystemsdc/iris-community:latest", port: int = 1972, |
| 31 | + username: Optional[str] = None, password: Optional[str] = None, |
| 32 | + dbname: Optional[str] = None, driver: str = "iris", **kwargs) -> None: |
| 33 | + # raise_for_deprecated_parameter(kwargs, "user", "username") |
| 34 | + super(IRISContainer, self).__init__(image=image, **kwargs) |
| 35 | + self.username = username or os.environ.get("IRIS_USERNAME", "test") |
| 36 | + self.password = password or os.environ.get("IRIS_PASSWORD", "test") |
| 37 | + self.dbname = dbname or os.environ.get("IRIS_NAMESPACE", "USER") |
| 38 | + self.port = port |
| 39 | + self.driver = driver |
| 40 | + |
| 41 | + self.with_exposed_ports(self.port) |
| 42 | + |
| 43 | + def _configure(self) -> None: |
| 44 | + self.with_env("IRIS_USERNAME", self.username) |
| 45 | + self.with_env("IRIS_PASSWORD", self.password) |
| 46 | + self.with_env("IRIS_NAMESPACE", self.dbname) |
| 47 | + |
| 48 | + def get_connection_url(self, host=None) -> str: |
| 49 | + return super()._create_connection_url( |
| 50 | + dialect=f"iris", username=self.username, |
| 51 | + password=self.password, db_name=self.dbname, host=host, |
| 52 | + port=self.port, |
| 53 | + ) |
0 commit comments