Configuration object for Postgres DB
| 660 | |
| 661 | |
| 662 | class PostgresConfig: |
| 663 | |
| 664 | """Configuration object for Postgres DB""" |
| 665 | |
| 666 | _conf = {"host": os.environ.get("USER_MANAGED_PG_HOST", "localhost"), |
| 667 | "port": os.environ.get("USER_MANAGED_PG_PORT", 5432), |
| 668 | "db_name": os.environ.get("USER_MANAGED_PG_DB_NAME", "postgres"), |
| 669 | "user_name": os.environ.get("USER_MANAGED_PG_USER_NAME", "postgres"), |
| 670 | "pw": os.environ.get("USER_MANAGED_PG_PW", ""), |
| 671 | # to create full copy, set "postgres_schema" to "full" |
| 672 | "pgvector_schema": "vector_only"} |
| 673 | |
| 674 | @classmethod |
| 675 | def get_config(cls, name): |
| 676 | if name in cls._conf: |
| 677 | return cls._conf[name] |
| 678 | raise ConfigKeyException(name) |
| 679 | |
| 680 | @classmethod |
| 681 | def set_config(cls, name, value): |
| 682 | cls._conf[name] = value |
| 683 | |
| 684 | @classmethod |
| 685 | def get_uri_string(cls): |
| 686 | |
| 687 | port = cls._conf["port"] |
| 688 | host = cls._conf["host"] |
| 689 | db_name = cls._conf["db_name"] |
| 690 | |
| 691 | # canonical simple format of postgres uri string |
| 692 | input_collection_db_path = f"postgresql://postgres@{host}:{port}/{db_name}" |
| 693 | # print("update: postgres get_uri_string - ", input_collection_db_path) |
| 694 | |
| 695 | return input_collection_db_path |
| 696 | |
| 697 | @classmethod |
| 698 | def get_db_configs(cls): |
| 699 | configs = {} |
| 700 | for keys, values in cls._conf.items(): |
| 701 | configs.update({keys:values}) |
| 702 | return configs |
| 703 | |
| 704 | @classmethod |
| 705 | def get_user_name(cls): |
| 706 | return cls._conf["user_name"] |
| 707 | |
| 708 | @classmethod |
| 709 | def get_db_pw(cls): |
| 710 | return cls._conf["pw"] |
| 711 | |
| 712 | |
| 713 | class RedisConfig: |