(self, config: PostgresDBConfig = PostgresDBConfig())
| 62 | |
| 63 | class PostgresDB(VectorStore): |
| 64 | def __init__(self, config: PostgresDBConfig = PostgresDBConfig()): |
| 65 | super().__init__(config) |
| 66 | if not has_postgres: |
| 67 | raise LangroidImportError("pgvector", "postgres") |
| 68 | try: |
| 69 | from sqlalchemy.orm import sessionmaker |
| 70 | except ImportError: |
| 71 | raise LangroidImportError("sqlalchemy", "postgres") |
| 72 | |
| 73 | self.config: PostgresDBConfig = config |
| 74 | self.engine = self._create_engine() |
| 75 | PostgresDB._create_vector_extension(self.engine) |
| 76 | self.SessionLocal = sessionmaker( |
| 77 | autocommit=False, autoflush=False, bind=self.engine |
| 78 | ) |
| 79 | self.metadata = MetaData() |
| 80 | self._setup_table() |
| 81 | |
| 82 | def _create_engine(self) -> Engine: |
| 83 | """Creates a SQLAlchemy engine based on the configuration.""" |
nothing calls this directly
no test coverage detected