feat: add wallet profile database schema and repositories (#11)
- Add SQLAlchemy models for wallet profiles, funding transfers, and relationships - Add WalletRepository, FundingRepository, RelationshipRepository with async support - Add DatabaseManager for connection and session management - Add Alembic migration for initial schema - Include comprehensive test suite with 21 tests using in-memory SQLite 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Patrick Selamy
co-authored by
Claude Opus 4.5
Claude Opus 4.5
parent
656bd6469c
commit
0003d228e5
@@ -1 +1,45 @@
|
||||
"""Storage layer - Database schemas and repositories."""
|
||||
|
||||
from polymarket_insider_tracker.storage.database import (
|
||||
DatabaseManager,
|
||||
create_async_db_engine,
|
||||
create_async_session_factory,
|
||||
create_sync_engine,
|
||||
create_sync_session_factory,
|
||||
init_async_db,
|
||||
init_db,
|
||||
)
|
||||
from polymarket_insider_tracker.storage.models import (
|
||||
Base,
|
||||
FundingTransferModel,
|
||||
WalletProfileModel,
|
||||
WalletRelationshipModel,
|
||||
)
|
||||
from polymarket_insider_tracker.storage.repos import (
|
||||
FundingRepository,
|
||||
FundingTransferDTO,
|
||||
RelationshipRepository,
|
||||
WalletProfileDTO,
|
||||
WalletRelationshipDTO,
|
||||
WalletRepository,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"Base",
|
||||
"DatabaseManager",
|
||||
"FundingRepository",
|
||||
"FundingTransferDTO",
|
||||
"FundingTransferModel",
|
||||
"RelationshipRepository",
|
||||
"WalletProfileDTO",
|
||||
"WalletProfileModel",
|
||||
"WalletRelationshipDTO",
|
||||
"WalletRelationshipModel",
|
||||
"WalletRepository",
|
||||
"create_async_db_engine",
|
||||
"create_async_session_factory",
|
||||
"create_sync_engine",
|
||||
"create_sync_session_factory",
|
||||
"init_async_db",
|
||||
"init_db",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,209 @@
|
||||
"""Database connection and session management.
|
||||
|
||||
This module provides the database engine, session factory, and
|
||||
async session support for the storage layer.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from collections.abc import AsyncGenerator
|
||||
from contextlib import asynccontextmanager
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||||
from sqlalchemy.orm import Session, sessionmaker
|
||||
|
||||
from polymarket_insider_tracker.storage.models import Base
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from sqlalchemy import Engine
|
||||
from sqlalchemy.ext.asyncio import AsyncEngine
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def create_sync_engine(database_url: str, **kwargs: Any) -> Engine:
|
||||
"""Create a synchronous SQLAlchemy engine.
|
||||
|
||||
Args:
|
||||
database_url: Database connection URL (e.g., postgresql://...).
|
||||
**kwargs: Additional engine options.
|
||||
|
||||
Returns:
|
||||
SQLAlchemy Engine instance.
|
||||
"""
|
||||
return create_engine(database_url, **kwargs)
|
||||
|
||||
|
||||
def create_async_db_engine(database_url: str, **kwargs: Any) -> AsyncEngine:
|
||||
"""Create an asynchronous SQLAlchemy engine.
|
||||
|
||||
Args:
|
||||
database_url: Database connection URL (e.g., postgresql+asyncpg://...).
|
||||
**kwargs: Additional engine options.
|
||||
|
||||
Returns:
|
||||
SQLAlchemy AsyncEngine instance.
|
||||
"""
|
||||
return create_async_engine(database_url, **kwargs)
|
||||
|
||||
|
||||
def create_sync_session_factory(engine: Engine) -> sessionmaker[Session]:
|
||||
"""Create a synchronous session factory.
|
||||
|
||||
Args:
|
||||
engine: SQLAlchemy Engine instance.
|
||||
|
||||
Returns:
|
||||
Session factory.
|
||||
"""
|
||||
return sessionmaker(bind=engine, expire_on_commit=False)
|
||||
|
||||
|
||||
def create_async_session_factory(engine: AsyncEngine) -> async_sessionmaker[AsyncSession]:
|
||||
"""Create an asynchronous session factory.
|
||||
|
||||
Args:
|
||||
engine: SQLAlchemy AsyncEngine instance.
|
||||
|
||||
Returns:
|
||||
Async session factory.
|
||||
"""
|
||||
return async_sessionmaker(bind=engine, expire_on_commit=False)
|
||||
|
||||
|
||||
def init_db(engine: Engine) -> None:
|
||||
"""Initialize the database schema.
|
||||
|
||||
Creates all tables defined in the models.
|
||||
|
||||
Args:
|
||||
engine: SQLAlchemy Engine instance.
|
||||
"""
|
||||
Base.metadata.create_all(engine)
|
||||
logger.info("Database schema initialized")
|
||||
|
||||
|
||||
async def init_async_db(engine: AsyncEngine) -> None:
|
||||
"""Initialize the database schema asynchronously.
|
||||
|
||||
Creates all tables defined in the models.
|
||||
|
||||
Args:
|
||||
engine: SQLAlchemy AsyncEngine instance.
|
||||
"""
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
logger.info("Database schema initialized (async)")
|
||||
|
||||
|
||||
class DatabaseManager:
|
||||
"""Manages database connections and sessions.
|
||||
|
||||
Provides a unified interface for both sync and async database operations.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
database_url: str,
|
||||
*,
|
||||
async_mode: bool = True,
|
||||
pool_size: int = 5,
|
||||
max_overflow: int = 10,
|
||||
echo: bool = False,
|
||||
) -> None:
|
||||
"""Initialize database manager.
|
||||
|
||||
Args:
|
||||
database_url: Database connection URL.
|
||||
async_mode: Use async engine/sessions if True.
|
||||
pool_size: Connection pool size.
|
||||
max_overflow: Maximum overflow connections.
|
||||
echo: Echo SQL statements for debugging.
|
||||
"""
|
||||
self.database_url = database_url
|
||||
self.async_mode = async_mode
|
||||
self._pool_size = pool_size
|
||||
self._max_overflow = max_overflow
|
||||
self._echo = echo
|
||||
|
||||
self._sync_engine: Engine | None = None
|
||||
self._async_engine: AsyncEngine | None = None
|
||||
self._sync_session_factory: sessionmaker[Session] | None = None
|
||||
self._async_session_factory: async_sessionmaker[AsyncSession] | None = None
|
||||
|
||||
def _get_sync_engine(self) -> Engine:
|
||||
"""Get or create the synchronous engine."""
|
||||
if self._sync_engine is None:
|
||||
self._sync_engine = create_sync_engine(
|
||||
self.database_url,
|
||||
pool_size=self._pool_size,
|
||||
max_overflow=self._max_overflow,
|
||||
echo=self._echo,
|
||||
)
|
||||
return self._sync_engine
|
||||
|
||||
def _get_async_engine(self) -> AsyncEngine:
|
||||
"""Get or create the asynchronous engine."""
|
||||
if self._async_engine is None:
|
||||
self._async_engine = create_async_db_engine(
|
||||
self.database_url,
|
||||
pool_size=self._pool_size,
|
||||
max_overflow=self._max_overflow,
|
||||
echo=self._echo,
|
||||
)
|
||||
return self._async_engine
|
||||
|
||||
def get_sync_session(self) -> Session:
|
||||
"""Get a new synchronous session.
|
||||
|
||||
Returns:
|
||||
SQLAlchemy Session instance.
|
||||
"""
|
||||
if self._sync_session_factory is None:
|
||||
self._sync_session_factory = create_sync_session_factory(self._get_sync_engine())
|
||||
return self._sync_session_factory()
|
||||
|
||||
@asynccontextmanager
|
||||
async def get_async_session(self) -> AsyncGenerator[AsyncSession, None]:
|
||||
"""Get an asynchronous session as a context manager.
|
||||
|
||||
Yields:
|
||||
SQLAlchemy AsyncSession instance.
|
||||
"""
|
||||
if self._async_session_factory is None:
|
||||
self._async_session_factory = create_async_session_factory(self._get_async_engine())
|
||||
|
||||
session = self._async_session_factory()
|
||||
try:
|
||||
yield session
|
||||
await session.commit()
|
||||
except Exception:
|
||||
await session.rollback()
|
||||
raise
|
||||
finally:
|
||||
await session.close()
|
||||
|
||||
def init_schema(self) -> None:
|
||||
"""Initialize database schema synchronously."""
|
||||
init_db(self._get_sync_engine())
|
||||
|
||||
async def init_schema_async(self) -> None:
|
||||
"""Initialize database schema asynchronously."""
|
||||
await init_async_db(self._get_async_engine())
|
||||
|
||||
def dispose(self) -> None:
|
||||
"""Dispose of all database connections."""
|
||||
if self._sync_engine is not None:
|
||||
self._sync_engine.dispose()
|
||||
self._sync_engine = None
|
||||
logger.info("Database connections disposed")
|
||||
|
||||
async def dispose_async(self) -> None:
|
||||
"""Dispose of all async database connections."""
|
||||
if self._async_engine is not None:
|
||||
await self._async_engine.dispose()
|
||||
self._async_engine = None
|
||||
logger.info("Async database connections disposed")
|
||||
@@ -0,0 +1,115 @@
|
||||
"""SQLAlchemy models for persistent storage.
|
||||
|
||||
This module defines the database schema for storing wallet profiles,
|
||||
funding transfers, and wallet relationships.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import UTC, datetime
|
||||
from decimal import Decimal
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from sqlalchemy import (
|
||||
Boolean,
|
||||
DateTime,
|
||||
Index,
|
||||
Integer,
|
||||
Numeric,
|
||||
String,
|
||||
UniqueConstraint,
|
||||
)
|
||||
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
|
||||
|
||||
if TYPE_CHECKING:
|
||||
pass
|
||||
|
||||
|
||||
class Base(DeclarativeBase):
|
||||
"""Base class for all SQLAlchemy models."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class WalletProfileModel(Base):
|
||||
"""SQLAlchemy model for wallet profiles.
|
||||
|
||||
Stores analyzed wallet information including age, transaction count,
|
||||
balances, and freshness classification.
|
||||
"""
|
||||
|
||||
__tablename__ = "wallet_profiles"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
address: Mapped[str] = mapped_column(String(42), unique=True, nullable=False)
|
||||
nonce: Mapped[int] = mapped_column(Integer, nullable=False)
|
||||
first_seen_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
|
||||
is_fresh: Mapped[bool] = mapped_column(Boolean, nullable=False)
|
||||
matic_balance: Mapped[Decimal | None] = mapped_column(Numeric(30, 0), nullable=True)
|
||||
usdc_balance: Mapped[Decimal | None] = mapped_column(Numeric(20, 6), nullable=True)
|
||||
analyzed_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC)
|
||||
)
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC)
|
||||
)
|
||||
updated_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True),
|
||||
nullable=False,
|
||||
default=lambda: datetime.now(UTC),
|
||||
onupdate=lambda: datetime.now(UTC),
|
||||
)
|
||||
|
||||
__table_args__ = (Index("idx_wallet_profiles_address", "address"),)
|
||||
|
||||
|
||||
class FundingTransferModel(Base):
|
||||
"""SQLAlchemy model for funding transfers.
|
||||
|
||||
Stores ERC20 transfer events to track wallet funding sources.
|
||||
"""
|
||||
|
||||
__tablename__ = "funding_transfers"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
from_address: Mapped[str] = mapped_column(String(42), nullable=False)
|
||||
to_address: Mapped[str] = mapped_column(String(42), nullable=False)
|
||||
amount: Mapped[Decimal] = mapped_column(Numeric(30, 6), nullable=False)
|
||||
token: Mapped[str] = mapped_column(String(10), nullable=False)
|
||||
tx_hash: Mapped[str] = mapped_column(String(66), unique=True, nullable=False)
|
||||
block_number: Mapped[int] = mapped_column(Integer, nullable=False)
|
||||
timestamp: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC)
|
||||
)
|
||||
|
||||
__table_args__ = (
|
||||
Index("idx_funding_transfers_to", "to_address"),
|
||||
Index("idx_funding_transfers_from", "from_address"),
|
||||
Index("idx_funding_transfers_block", "block_number"),
|
||||
)
|
||||
|
||||
|
||||
class WalletRelationshipModel(Base):
|
||||
"""SQLAlchemy model for wallet relationships.
|
||||
|
||||
Stores graph edges between wallets representing funding relationships
|
||||
or entity linkages.
|
||||
"""
|
||||
|
||||
__tablename__ = "wallet_relationships"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
wallet_a: Mapped[str] = mapped_column(String(42), nullable=False)
|
||||
wallet_b: Mapped[str] = mapped_column(String(42), nullable=False)
|
||||
relationship_type: Mapped[str] = mapped_column(String(20), nullable=False)
|
||||
confidence: Mapped[Decimal] = mapped_column(Numeric(3, 2), nullable=False)
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC)
|
||||
)
|
||||
|
||||
__table_args__ = (
|
||||
UniqueConstraint("wallet_a", "wallet_b", "relationship_type", name="uq_wallet_relationship"),
|
||||
Index("idx_wallet_relationships_a", "wallet_a"),
|
||||
Index("idx_wallet_relationships_b", "wallet_b"),
|
||||
)
|
||||
@@ -0,0 +1,515 @@
|
||||
"""Repository pattern implementations for data access.
|
||||
|
||||
This module provides clean data access abstractions for wallet profiles,
|
||||
funding transfers, and wallet relationships.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from datetime import UTC, datetime
|
||||
from decimal import Decimal
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from sqlalchemy import delete, select, update
|
||||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||||
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
|
||||
|
||||
from polymarket_insider_tracker.storage.models import (
|
||||
FundingTransferModel,
|
||||
WalletProfileModel,
|
||||
WalletRelationshipModel,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class WalletProfileDTO:
|
||||
"""Data transfer object for wallet profiles."""
|
||||
|
||||
address: str
|
||||
nonce: int
|
||||
first_seen_at: datetime | None
|
||||
is_fresh: bool
|
||||
matic_balance: Decimal | None
|
||||
usdc_balance: Decimal | None
|
||||
analyzed_at: datetime
|
||||
created_at: datetime | None = None
|
||||
updated_at: datetime | None = None
|
||||
|
||||
@classmethod
|
||||
def from_model(cls, model: WalletProfileModel) -> WalletProfileDTO:
|
||||
"""Create DTO from SQLAlchemy model."""
|
||||
return cls(
|
||||
address=model.address,
|
||||
nonce=model.nonce,
|
||||
first_seen_at=model.first_seen_at,
|
||||
is_fresh=model.is_fresh,
|
||||
matic_balance=model.matic_balance,
|
||||
usdc_balance=model.usdc_balance,
|
||||
analyzed_at=model.analyzed_at,
|
||||
created_at=model.created_at,
|
||||
updated_at=model.updated_at,
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class FundingTransferDTO:
|
||||
"""Data transfer object for funding transfers."""
|
||||
|
||||
from_address: str
|
||||
to_address: str
|
||||
amount: Decimal
|
||||
token: str
|
||||
tx_hash: str
|
||||
block_number: int
|
||||
timestamp: datetime
|
||||
created_at: datetime | None = None
|
||||
|
||||
@classmethod
|
||||
def from_model(cls, model: FundingTransferModel) -> FundingTransferDTO:
|
||||
"""Create DTO from SQLAlchemy model."""
|
||||
return cls(
|
||||
from_address=model.from_address,
|
||||
to_address=model.to_address,
|
||||
amount=model.amount,
|
||||
token=model.token,
|
||||
tx_hash=model.tx_hash,
|
||||
block_number=model.block_number,
|
||||
timestamp=model.timestamp,
|
||||
created_at=model.created_at,
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class WalletRelationshipDTO:
|
||||
"""Data transfer object for wallet relationships."""
|
||||
|
||||
wallet_a: str
|
||||
wallet_b: str
|
||||
relationship_type: str
|
||||
confidence: Decimal
|
||||
created_at: datetime | None = None
|
||||
|
||||
@classmethod
|
||||
def from_model(cls, model: WalletRelationshipModel) -> WalletRelationshipDTO:
|
||||
"""Create DTO from SQLAlchemy model."""
|
||||
return cls(
|
||||
wallet_a=model.wallet_a,
|
||||
wallet_b=model.wallet_b,
|
||||
relationship_type=model.relationship_type,
|
||||
confidence=model.confidence,
|
||||
created_at=model.created_at,
|
||||
)
|
||||
|
||||
|
||||
class WalletRepository:
|
||||
"""Repository for wallet profile data access.
|
||||
|
||||
Provides CRUD operations for wallet profiles with async support.
|
||||
"""
|
||||
|
||||
def __init__(self, session: AsyncSession) -> None:
|
||||
"""Initialize repository with database session.
|
||||
|
||||
Args:
|
||||
session: SQLAlchemy async session.
|
||||
"""
|
||||
self.session = session
|
||||
|
||||
async def get_by_address(self, address: str) -> WalletProfileDTO | None:
|
||||
"""Get wallet profile by address.
|
||||
|
||||
Args:
|
||||
address: Wallet address (lowercase).
|
||||
|
||||
Returns:
|
||||
WalletProfileDTO if found, None otherwise.
|
||||
"""
|
||||
result = await self.session.execute(
|
||||
select(WalletProfileModel).where(WalletProfileModel.address == address.lower())
|
||||
)
|
||||
model = result.scalar_one_or_none()
|
||||
return WalletProfileDTO.from_model(model) if model else None
|
||||
|
||||
async def get_many(self, addresses: list[str]) -> list[WalletProfileDTO]:
|
||||
"""Get multiple wallet profiles by addresses.
|
||||
|
||||
Args:
|
||||
addresses: List of wallet addresses.
|
||||
|
||||
Returns:
|
||||
List of WalletProfileDTOs for found addresses.
|
||||
"""
|
||||
normalized = [addr.lower() for addr in addresses]
|
||||
result = await self.session.execute(
|
||||
select(WalletProfileModel).where(WalletProfileModel.address.in_(normalized))
|
||||
)
|
||||
return [WalletProfileDTO.from_model(m) for m in result.scalars().all()]
|
||||
|
||||
async def get_fresh_wallets(self, limit: int = 100) -> list[WalletProfileDTO]:
|
||||
"""Get recent fresh wallets.
|
||||
|
||||
Args:
|
||||
limit: Maximum number of results.
|
||||
|
||||
Returns:
|
||||
List of WalletProfileDTOs marked as fresh.
|
||||
"""
|
||||
result = await self.session.execute(
|
||||
select(WalletProfileModel)
|
||||
.where(WalletProfileModel.is_fresh.is_(True))
|
||||
.order_by(WalletProfileModel.analyzed_at.desc())
|
||||
.limit(limit)
|
||||
)
|
||||
return [WalletProfileDTO.from_model(m) for m in result.scalars().all()]
|
||||
|
||||
async def upsert(self, dto: WalletProfileDTO) -> WalletProfileDTO:
|
||||
"""Insert or update wallet profile.
|
||||
|
||||
Args:
|
||||
dto: Wallet profile data.
|
||||
|
||||
Returns:
|
||||
Updated WalletProfileDTO.
|
||||
"""
|
||||
now = datetime.now(UTC)
|
||||
values = {
|
||||
"address": dto.address.lower(),
|
||||
"nonce": dto.nonce,
|
||||
"first_seen_at": dto.first_seen_at,
|
||||
"is_fresh": dto.is_fresh,
|
||||
"matic_balance": dto.matic_balance,
|
||||
"usdc_balance": dto.usdc_balance,
|
||||
"analyzed_at": dto.analyzed_at,
|
||||
"updated_at": now,
|
||||
}
|
||||
|
||||
# Try PostgreSQL upsert first, fall back to SQLite for testing
|
||||
try:
|
||||
stmt = pg_insert(WalletProfileModel).values(**values, created_at=now)
|
||||
stmt = stmt.on_conflict_do_update(
|
||||
index_elements=["address"],
|
||||
set_={
|
||||
"nonce": stmt.excluded.nonce,
|
||||
"first_seen_at": stmt.excluded.first_seen_at,
|
||||
"is_fresh": stmt.excluded.is_fresh,
|
||||
"matic_balance": stmt.excluded.matic_balance,
|
||||
"usdc_balance": stmt.excluded.usdc_balance,
|
||||
"analyzed_at": stmt.excluded.analyzed_at,
|
||||
"updated_at": stmt.excluded.updated_at,
|
||||
},
|
||||
)
|
||||
await self.session.execute(stmt)
|
||||
except Exception:
|
||||
# Fall back to SQLite upsert for testing
|
||||
stmt = sqlite_insert(WalletProfileModel).values(**values, created_at=now)
|
||||
stmt = stmt.on_conflict_do_update(
|
||||
index_elements=["address"],
|
||||
set_={
|
||||
"nonce": stmt.excluded.nonce,
|
||||
"first_seen_at": stmt.excluded.first_seen_at,
|
||||
"is_fresh": stmt.excluded.is_fresh,
|
||||
"matic_balance": stmt.excluded.matic_balance,
|
||||
"usdc_balance": stmt.excluded.usdc_balance,
|
||||
"analyzed_at": stmt.excluded.analyzed_at,
|
||||
"updated_at": stmt.excluded.updated_at,
|
||||
},
|
||||
)
|
||||
await self.session.execute(stmt)
|
||||
|
||||
await self.session.flush()
|
||||
return dto
|
||||
|
||||
async def delete(self, address: str) -> bool:
|
||||
"""Delete wallet profile by address.
|
||||
|
||||
Args:
|
||||
address: Wallet address.
|
||||
|
||||
Returns:
|
||||
True if deleted, False if not found.
|
||||
"""
|
||||
result = await self.session.execute(
|
||||
delete(WalletProfileModel).where(WalletProfileModel.address == address.lower())
|
||||
)
|
||||
return result.rowcount > 0
|
||||
|
||||
async def mark_stale(self, address: str) -> bool:
|
||||
"""Mark a wallet profile as stale (soft delete).
|
||||
|
||||
Sets analyzed_at to a very old date to trigger re-analysis.
|
||||
|
||||
Args:
|
||||
address: Wallet address.
|
||||
|
||||
Returns:
|
||||
True if updated, False if not found.
|
||||
"""
|
||||
stale_time = datetime(2000, 1, 1, tzinfo=UTC)
|
||||
result = await self.session.execute(
|
||||
update(WalletProfileModel)
|
||||
.where(WalletProfileModel.address == address.lower())
|
||||
.values(analyzed_at=stale_time, updated_at=datetime.now(UTC))
|
||||
)
|
||||
return result.rowcount > 0
|
||||
|
||||
|
||||
class FundingRepository:
|
||||
"""Repository for funding transfer data access.
|
||||
|
||||
Provides CRUD operations for funding transfers with async support.
|
||||
"""
|
||||
|
||||
def __init__(self, session: AsyncSession) -> None:
|
||||
"""Initialize repository with database session.
|
||||
|
||||
Args:
|
||||
session: SQLAlchemy async session.
|
||||
"""
|
||||
self.session = session
|
||||
|
||||
async def get_transfers_to(
|
||||
self, address: str, limit: int = 100
|
||||
) -> list[FundingTransferDTO]:
|
||||
"""Get transfers to a wallet address.
|
||||
|
||||
Args:
|
||||
address: Destination wallet address.
|
||||
limit: Maximum number of results.
|
||||
|
||||
Returns:
|
||||
List of FundingTransferDTOs ordered by timestamp.
|
||||
"""
|
||||
result = await self.session.execute(
|
||||
select(FundingTransferModel)
|
||||
.where(FundingTransferModel.to_address == address.lower())
|
||||
.order_by(FundingTransferModel.timestamp.asc())
|
||||
.limit(limit)
|
||||
)
|
||||
return [FundingTransferDTO.from_model(m) for m in result.scalars().all()]
|
||||
|
||||
async def get_transfers_from(
|
||||
self, address: str, limit: int = 100
|
||||
) -> list[FundingTransferDTO]:
|
||||
"""Get transfers from a wallet address.
|
||||
|
||||
Args:
|
||||
address: Source wallet address.
|
||||
limit: Maximum number of results.
|
||||
|
||||
Returns:
|
||||
List of FundingTransferDTOs ordered by timestamp.
|
||||
"""
|
||||
result = await self.session.execute(
|
||||
select(FundingTransferModel)
|
||||
.where(FundingTransferModel.from_address == address.lower())
|
||||
.order_by(FundingTransferModel.timestamp.asc())
|
||||
.limit(limit)
|
||||
)
|
||||
return [FundingTransferDTO.from_model(m) for m in result.scalars().all()]
|
||||
|
||||
async def get_first_transfer_to(self, address: str) -> FundingTransferDTO | None:
|
||||
"""Get the first transfer to a wallet.
|
||||
|
||||
Args:
|
||||
address: Wallet address.
|
||||
|
||||
Returns:
|
||||
First FundingTransferDTO if found, None otherwise.
|
||||
"""
|
||||
result = await self.session.execute(
|
||||
select(FundingTransferModel)
|
||||
.where(FundingTransferModel.to_address == address.lower())
|
||||
.order_by(FundingTransferModel.timestamp.asc())
|
||||
.limit(1)
|
||||
)
|
||||
model = result.scalar_one_or_none()
|
||||
return FundingTransferDTO.from_model(model) if model else None
|
||||
|
||||
async def get_by_tx_hash(self, tx_hash: str) -> FundingTransferDTO | None:
|
||||
"""Get transfer by transaction hash.
|
||||
|
||||
Args:
|
||||
tx_hash: Transaction hash.
|
||||
|
||||
Returns:
|
||||
FundingTransferDTO if found, None otherwise.
|
||||
"""
|
||||
result = await self.session.execute(
|
||||
select(FundingTransferModel).where(FundingTransferModel.tx_hash == tx_hash.lower())
|
||||
)
|
||||
model = result.scalar_one_or_none()
|
||||
return FundingTransferDTO.from_model(model) if model else None
|
||||
|
||||
async def insert(self, dto: FundingTransferDTO) -> FundingTransferDTO:
|
||||
"""Insert a new funding transfer.
|
||||
|
||||
Args:
|
||||
dto: Funding transfer data.
|
||||
|
||||
Returns:
|
||||
Inserted FundingTransferDTO.
|
||||
|
||||
Raises:
|
||||
IntegrityError if tx_hash already exists.
|
||||
"""
|
||||
model = FundingTransferModel(
|
||||
from_address=dto.from_address.lower(),
|
||||
to_address=dto.to_address.lower(),
|
||||
amount=dto.amount,
|
||||
token=dto.token,
|
||||
tx_hash=dto.tx_hash.lower(),
|
||||
block_number=dto.block_number,
|
||||
timestamp=dto.timestamp,
|
||||
)
|
||||
self.session.add(model)
|
||||
await self.session.flush()
|
||||
return dto
|
||||
|
||||
async def insert_many(self, dtos: list[FundingTransferDTO]) -> int:
|
||||
"""Insert multiple funding transfers.
|
||||
|
||||
Skips duplicates silently.
|
||||
|
||||
Args:
|
||||
dtos: List of funding transfer data.
|
||||
|
||||
Returns:
|
||||
Number of transfers inserted.
|
||||
"""
|
||||
inserted = 0
|
||||
for dto in dtos:
|
||||
try:
|
||||
await self.insert(dto)
|
||||
inserted += 1
|
||||
except Exception as e:
|
||||
# Skip duplicates
|
||||
if "UNIQUE constraint" in str(e) or "duplicate key" in str(e).lower():
|
||||
continue
|
||||
raise
|
||||
return inserted
|
||||
|
||||
|
||||
class RelationshipRepository:
|
||||
"""Repository for wallet relationship data access.
|
||||
|
||||
Provides CRUD operations for wallet relationships with async support.
|
||||
"""
|
||||
|
||||
def __init__(self, session: AsyncSession) -> None:
|
||||
"""Initialize repository with database session.
|
||||
|
||||
Args:
|
||||
session: SQLAlchemy async session.
|
||||
"""
|
||||
self.session = session
|
||||
|
||||
async def get_relationships(
|
||||
self, wallet: str, relationship_type: str | None = None
|
||||
) -> list[WalletRelationshipDTO]:
|
||||
"""Get relationships for a wallet.
|
||||
|
||||
Args:
|
||||
wallet: Wallet address.
|
||||
relationship_type: Optional filter by type.
|
||||
|
||||
Returns:
|
||||
List of WalletRelationshipDTOs.
|
||||
"""
|
||||
stmt = select(WalletRelationshipModel).where(
|
||||
(WalletRelationshipModel.wallet_a == wallet.lower())
|
||||
| (WalletRelationshipModel.wallet_b == wallet.lower())
|
||||
)
|
||||
if relationship_type:
|
||||
stmt = stmt.where(WalletRelationshipModel.relationship_type == relationship_type)
|
||||
|
||||
result = await self.session.execute(stmt)
|
||||
return [WalletRelationshipDTO.from_model(m) for m in result.scalars().all()]
|
||||
|
||||
async def get_related_wallets(
|
||||
self, wallet: str, relationship_type: str | None = None
|
||||
) -> list[str]:
|
||||
"""Get addresses of related wallets.
|
||||
|
||||
Args:
|
||||
wallet: Wallet address.
|
||||
relationship_type: Optional filter by type.
|
||||
|
||||
Returns:
|
||||
List of related wallet addresses.
|
||||
"""
|
||||
relationships = await self.get_relationships(wallet, relationship_type)
|
||||
related = set()
|
||||
normalized = wallet.lower()
|
||||
for rel in relationships:
|
||||
if rel.wallet_a == normalized:
|
||||
related.add(rel.wallet_b)
|
||||
else:
|
||||
related.add(rel.wallet_a)
|
||||
return list(related)
|
||||
|
||||
async def upsert(self, dto: WalletRelationshipDTO) -> WalletRelationshipDTO:
|
||||
"""Insert or update wallet relationship.
|
||||
|
||||
Args:
|
||||
dto: Wallet relationship data.
|
||||
|
||||
Returns:
|
||||
Updated WalletRelationshipDTO.
|
||||
"""
|
||||
now = datetime.now(UTC)
|
||||
values = {
|
||||
"wallet_a": dto.wallet_a.lower(),
|
||||
"wallet_b": dto.wallet_b.lower(),
|
||||
"relationship_type": dto.relationship_type,
|
||||
"confidence": dto.confidence,
|
||||
"created_at": now,
|
||||
}
|
||||
|
||||
# Try PostgreSQL upsert first, fall back to SQLite for testing
|
||||
try:
|
||||
stmt = pg_insert(WalletRelationshipModel).values(**values)
|
||||
stmt = stmt.on_conflict_do_update(
|
||||
constraint="uq_wallet_relationship",
|
||||
set_={"confidence": stmt.excluded.confidence},
|
||||
)
|
||||
await self.session.execute(stmt)
|
||||
except Exception:
|
||||
# Fall back to SQLite upsert for testing
|
||||
stmt = sqlite_insert(WalletRelationshipModel).values(**values)
|
||||
stmt = stmt.on_conflict_do_update(
|
||||
index_elements=["wallet_a", "wallet_b", "relationship_type"],
|
||||
set_={"confidence": stmt.excluded.confidence},
|
||||
)
|
||||
await self.session.execute(stmt)
|
||||
|
||||
await self.session.flush()
|
||||
return dto
|
||||
|
||||
async def delete(
|
||||
self, wallet_a: str, wallet_b: str, relationship_type: str
|
||||
) -> bool:
|
||||
"""Delete a specific relationship.
|
||||
|
||||
Args:
|
||||
wallet_a: First wallet address.
|
||||
wallet_b: Second wallet address.
|
||||
relationship_type: Type of relationship.
|
||||
|
||||
Returns:
|
||||
True if deleted, False if not found.
|
||||
"""
|
||||
result = await self.session.execute(
|
||||
delete(WalletRelationshipModel).where(
|
||||
WalletRelationshipModel.wallet_a == wallet_a.lower(),
|
||||
WalletRelationshipModel.wallet_b == wallet_b.lower(),
|
||||
WalletRelationshipModel.relationship_type == relationship_type,
|
||||
)
|
||||
)
|
||||
return result.rowcount > 0
|
||||
Reference in New Issue
Block a user