Database

PostgreSQLConfig

Bases: BaseSettings

Methods:

Name	Description
init_db

postgres_url

postgres_url: str = Field(
    default="postgresql://postgres:postgres@postgres:5432/postgres",
    validation_alias=AliasChoices("POSTGRES_URL"),
    title="PostgreSQL Url",
    description="The URL to connect to the PostgreSQL database.",
    frozen=False,
    deprecated=False,
)

init_db

init_db() -> None

Source code in src/types/database.py

def init_db(self) -> None:
    try:
        engine = create_engine(self.postgres_url)

        with engine.connect() as conn:
            conn.execute(text("SELECT 1"))
            logfire.info("PostgreSQL database connection successful")

    except SQLAlchemyError:
        try:
            postgres_url = self.postgres_url
            if "/" in postgres_url:
                base_url = postgres_url.rsplit("/", 1)[0]
                database_name = postgres_url.rsplit("/", 1)[1]
                admin_engine = create_engine(f"{base_url}/postgres")

                with admin_engine.connect() as admin_conn:
                    admin_conn.execute(text("COMMIT"))
                    admin_conn.execute(text(f"CREATE DATABASE {database_name}"))
                    logfire.info(f"Successfully created database: {database_name}")

                    test_engine = create_engine(postgres_url)
                    with test_engine.connect() as test_conn:
                        test_conn.execute(text("SELECT 1"))
                        logfire.info("New database connection verified")

        except SQLAlchemyError as create_error:
            logfire.error(f"Failed to create database: {create_error}", _exc_info=True)
    except Exception as e:
        logfire.error(f"Unexpected error ensuring database: {e}", _exc_info=True)

SQLiteConfig

Bases: BaseSettings

sqlite_file_path

sqlite_file_path: str = Field(
    default="sqlite:///data/messages.db",
    validation_alias=AliasChoices("SQLITE_FILE_PATH"),
    title="SQLite File Path",
    description="The file path to the SQLite database file.",
    frozen=False,
    deprecated=False,
)

sqlite_timeout

sqlite_timeout: int = Field(
    default=30,
    validation_alias=AliasChoices("SQLITE_TIMEOUT"),
    title="SQLite Timeout",
    description="The timeout duration (in seconds) for SQLite operations. Defaults to 30 seconds.",
    frozen=False,
    deprecated=False,
)

RedisConfig

Bases: BaseSettings

Methods:

Name	Description
hvalues
save
load
delete

redis_url

redis_url: str = Field(
    default="redis://redis:6379/0",
    validation_alias=AliasChoices("REDIS_URL"),
    title="Redis Url",
    description="The URL to connect to the Redis server.",
    frozen=False,
    deprecated=False,
)

redis_instance

redis_instance: Redis

hkeys

hkeys: list[str]

hvalues

hvalues(key: str) -> list[str]

Source code in src/types/database.py

def hvalues(self, key: str) -> list[str]:
    # 取得特定 hash key 中所有的值
    values: list[bytes] = self.redis_instance.hvals(key)
    return [val.decode("utf-8") for val in values]

save

save(key: str, data: dict[str, str]) -> dict[str, Any]

Source code in src/types/database.py

def save(self, key: str, data: dict[str, str]) -> dict[str, Any]:
    # 使用 hset 將 data 字典保存到指定的 hash key 中
    # data 格式例如: {"field1": "value1", "field2": "value2"}
    self.redis_instance.hset(key, mapping=data)  # type: ignore
    return data

load

load(key: str) -> dict[str, str]

Source code in src/types/database.py

def load(self, key: str) -> dict[str, str]:
    # 使用 hgetall 取得指定 hash key 中所有 field-value
    raw_data: dict[bytes, bytes] = self.redis_instance.hgetall(key)
    if not raw_data:
        return {}
    return {k.decode("utf-8"): v.decode("utf-8") for k, v in raw_data.items()}

delete

delete(key: str) -> None

Source code in src/types/database.py

def delete(self, key: str) -> None:
    # 刪除整個 hash key
    self.redis_instance.delete(key)

DatabaseConfig

Bases: BaseModel

postgres

postgres: PostgreSQLConfig = PostgreSQLConfig()

sqlite

sqlite: SQLiteConfig = SQLiteConfig()

redis

redis: RedisConfig = RedisConfig()