Repository: Chia-Network/chia-blockchain
File: chia/data_layer/data_store.py (Python)

Code before:
from chia.util.db_wrapper import SQLITE_MAX_VARIABLE_NUMBER, DBWrapper2
from chia.util.log_exceptions import log_exceptions
from chia.util.lru_cache import LRUCache

log = logging.getLogger(__name__)


# TODO: review exceptions for values that shouldn't be displayed
# TODO: pick exception types other than Exception

KeyOrValueId = int64

default_prefer_file_kv_blob_length: int = 4096


@dataclass
class DataStore:
    """A key/value store with the pairs being terminal nodes in a CLVM object tree."""

    db_wrapper: DBWrapper2
    recent_merkle_blobs: LRUCache[bytes32, MerkleBlob]
    merkle_blobs_path: Path
    key_value_blobs_path: Path
    unconfirmed_keys_values: defaultdict[bytes32, list[bytes32]] = field(
        default_factory=functools.partial(defaultdict, list)
    )
    prefer_db_kv_blob_length: int = default_prefer_file_kv_blob_length

    @classmethod
    @contextlib.asynccontextmanager
    async def managed(
        cls,
        database: Union[str, Path],
        merkle_blobs_path: Path,
        key_value_blobs_path: Path,
        uri: bool = False,
        sql_log_path: Optional[Path] = None,
        cache_capacity: int = 1,
        prefer_db_kv_blob_length: int = default_prefer_file_kv_blob_length,
    ) -> AsyncIterator[DataStore]:
        async with DBWrapper2.managed(
            database=database,
            uri=uri,
            journal_mode="WAL",
            # Setting to FULL despite other locations being configurable.  If there are
            # performance issues we can consider other the implications of other options.
            synchronous="FULL",
            # If foreign key checking gets turned off, please add corresponding check
            # methods and enable foreign key checking in the tests.
            foreign_keys=True,
            row_factory=aiosqlite.Row,

Review comment (suggestion):
```suggestion
    unconfirmed_keys_values: dict[bytes32, list[bytes32]] = field(default_factory=dict)
```

Code after:
from chia.util.log_exceptions import log_exceptions
from chia.util.lru_cache import LRUCache

log = logging.getLogger(__name__)


# TODO: review exceptions for values that shouldn't be displayed
# TODO: pick exception types other than Exception

KeyOrValueId = int64

default_prefer_file_kv_blob_length: int = 4096


@dataclass
class DataStore:
    """A key/value store with the pairs being terminal nodes in a CLVM object tree."""

    db_wrapper: DBWrapper2
    recent_merkle_blobs: LRUCache[bytes32, MerkleBlob]
    merkle_blobs_path: Path
    key_value_blobs_path: Path
    unconfirmed_keys_values: dict[bytes32, list[bytes32]] = field(default_factory=dict)
    prefer_db_kv_blob_length: int = default_prefer_file_kv_blob_length

    @classmethod
    @contextlib.asynccontextmanager
    async def managed(
        cls,
        database: Union[str, Path],
        merkle_blobs_path: Path,
        key_value_blobs_path: Path,
        uri: bool = False,
        sql_log_path: Optional[Path] = None,
        cache_capacity: int = 1,
        prefer_db_kv_blob_length: int = default_prefer_file_kv_blob_length,
    ) -> AsyncIterator[DataStore]:
        async with DBWrapper2.managed(
            database=database,
            uri=uri,
            journal_mode="WAL",
            # Setting to FULL despite other locations being configurable.  If there are
            # performance issues we can consider other the implications of other options.
            synchronous="FULL",
            # If foreign key checking gets turned off, please add corresponding check
            # methods and enable foreign key checking in the tests.
            foreign_keys=True,
            row_factory=aiosqlite.Row,
            log_path=sql_log_path,
        ) as db_wrapper:
            recent_merkle_blobs: LRUCache[bytes32, MerkleBlob] = LRUCache(capacity=cache_capacity)