Repository: Chia-Network/chia-blockchain
File: chia/full_node/coin_store.py (Python)

Code before:
from __future__ import annotations

import dataclasses
import logging
import sqlite3
import time
from collections.abc import Collection
from typing import Any, ClassVar, Optional

import typing_extensions
from aiosqlite import Cursor
from chia_rs import CoinState
from chia_rs.sized_bytes import bytes32
from chia_rs.sized_ints import uint32, uint64

from chia.types.blockchain_format.coin import Coin
from chia.types.coin_record import CoinRecord
from chia.types.mempool_item import UnspentLineageInfo
from chia.util.batches import to_batches
from chia.util.db_wrapper import SQLITE_MAX_VARIABLE_NUMBER, DBWrapper2

log = logging.getLogger(__name__)


async def add_coin_records_to_db(db_wrapper: DBWrapper2, records: list[CoinRecord]) -> None:
    if len(records) == 0:
        return
    async with db_wrapper.writer_maybe_transaction() as conn:
        await conn.executemany(
            "INSERT INTO coin_record VALUES(?, ?, ?, ?, ?, ?, ?, ?)",
            (
                (
                    record.coin.name(),
                    record.confirmed_block_index,
                    record.spent_block_index,
                    int(record.coinbase),
                    record.coin.puzzle_hash,
                    record.coin.parent_coin_info,
                    record.coin.amount.stream_to_bytes(),
                    record.timestamp,
                )
                for record in records
            ),
        )


@typing_extensions.final
@dataclasses.dataclass
class CoinStore:
    """

Review comment (suggestion):
I think most of the value of this change is gained by moving this function into a file that lives under `chia/_tests/` to remove it from production code and turn it into test code

Code after:
from __future__ import annotations

import dataclasses
import logging
import sqlite3
import time
from collections.abc import Collection
from typing import Any, ClassVar, Optional

import typing_extensions
from aiosqlite import Cursor
from chia_rs import CoinState
from chia_rs.sized_bytes import bytes32
from chia_rs.sized_ints import uint32, uint64

from chia.types.blockchain_format.coin import Coin
from chia.types.coin_record import CoinRecord
from chia.types.mempool_item import UnspentLineageInfo
from chia.util.batches import to_batches
from chia.util.db_wrapper import SQLITE_MAX_VARIABLE_NUMBER, DBWrapper2

log = logging.getLogger(__name__)


@typing_extensions.final
@dataclasses.dataclass
class CoinStore:
    """
    This object handles CoinRecords in DB.
    """

    db_wrapper: DBWrapper2

    @classmethod
    async def create(cls, db_wrapper: DBWrapper2) -> CoinStore:
        if db_wrapper.db_version != 2:
            raise RuntimeError(f"CoinStore does not support database schema v{db_wrapper.db_version}")
        self = CoinStore(db_wrapper)

        async with self.db_wrapper.writer_maybe_transaction() as conn:
            log.info("DB: Creating coin store tables and indexes.")
            # the coin_name is unique in this table because the CoinStore always
            # only represent a single peak
            await conn.execute(
                "CREATE TABLE IF NOT EXISTS coin_record("
                "coin_name blob PRIMARY KEY,"
                " confirmed_index bigint,"
                " spent_index bigint,"  # if this is zero, it means the coin has not been spent
                " coinbase int,"
                " puzzle_hash blob,"