Method get_instance

datastudio/utils/database.py:149–185

Get or create ShardedLMDBManager singleton. Args: cache_dir: Cache directory path. num_shards: Number of shards. map_size_per_shard: Maximum size per shard. readonly: Read-only mode flag. lock: File locking flag. logger

(
        cls,
        cache_dir: str,
        num_shards: int = 32,
        map_size_per_shard: int = 1024 * 1024 * 1024 * 1024,
        readonly: bool = False,
        lock: bool = True,
        logger=None,
        max_open_shards: int = 0,
    )

    @classmethod
    def get_instance(
        cls,
        cache_dir: str,
        num_shards: int = 32,
        map_size_per_shard: int = 1024 * 1024 * 1024 * 1024,
        readonly: bool = False,
        lock: bool = True,
        logger=None,
        max_open_shards: int = 0,
    ) -> "ShardedLMDBManager":
        """Get or create ShardedLMDBManager singleton.

        Args:
            cache_dir: Cache directory path.
            num_shards: Number of shards.
            map_size_per_shard: Maximum size per shard.
            readonly: Read-only mode flag.
            lock: File locking flag.
            logger: Logger instance.
            max_open_shards: Maximum open shards (0 = no limit).

        Returns:
            ShardedLMDBManager instance for the given cache_dir.
        """
        cache_dir = os.path.abspath(os.path.expanduser(cache_dir))

        if cache_dir not in cls._instances:
            cls._instances[cache_dir] = cls(
                cache_dir=cache_dir,
                num_shards=num_shards,
                map_size_per_shard=map_size_per_shard,
                readonly=readonly,
                lock=lock,
                logger=logger,
                max_open_shards=max_open_shards,
            )
        return cls._instances[cache_dir]

    def _get_shard_path(self, shard_id: int) -> str:
        """Get path to shard directory."""