Class SqlRegistry

sdk/python/feast/infra/registry/sql.py:284–1843

class SqlRegistry(CachingRegistry):
    def __init__(
        self,
        registry_config,
        project: str,
        repo_path: Optional[Path],
    ):
        assert registry_config is not None and isinstance(
            registry_config, SqlRegistryConfig
        ), "SqlRegistry needs a valid registry_config"

        self.registry_config = registry_config

        self.write_engine: Engine = create_engine(
            registry_config.path, **registry_config.sqlalchemy_config_kwargs
        )
        if registry_config.read_path:
            self.read_engine: Engine = create_engine(
                registry_config.read_path,
                **registry_config.sqlalchemy_config_kwargs,
            )
        else:
            self.read_engine = self.write_engine
        metadata.create_all(self.write_engine)
        self.thread_pool_executor_worker_count = (
            registry_config.thread_pool_executor_worker_count
        )
        self.purge_feast_metadata = registry_config.purge_feast_metadata
        self.enable_online_versioning = (
            registry_config.enable_online_feature_view_versioning
        )
        super().__init__(
            project=project,
            cache_ttl_seconds=registry_config.cache_ttl_seconds,
            cache_mode=registry_config.cache_mode,
        )
        # Sync feast_metadata to projects table
        # when purge_feast_metadata is set to True, Delete data from
        # feast_metadata table and list_project_metadata will not return any data
        self._sync_feast_metadata_to_projects_table()
        if not self.purge_feast_metadata:
            self._maybe_init_project_metadata(project)

    def _sync_feast_metadata_to_projects_table(self):
        feast_metadata_projects: dict = {}
        projects_set: set = []
        with self.read_engine.begin() as conn:
            stmt = select(feast_metadata).where(
                feast_metadata.c.metadata_key == FeastMetadataKeys.PROJECT_UUID.value
            )
            rows = conn.execute(stmt).all()
            for row in rows:
                feast_metadata_projects[row._mapping["project_id"]] = int(
                    row._mapping["last_updated_timestamp"]
                )

        if len(feast_metadata_projects) > 0:
            with self.read_engine.begin() as conn: