alien_py_cloud/alien_store/repositories/contract_repo.py

from sqlalchemy.ext.asyncio import AsyncSession
from alien_store.db.models.contract_store import ContractStore


class ContractRepository:
    """合同数据访问层"""

    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_by_store_id(self, store_id: int):
        """根据店铺id查询所有合同"""
        result = await self.db.execute(
            ContractStore.__table__.select().where(ContractStore.store_id == store_id)
        )
        # 返回列表[dict]，避免 Pydantic 序列化 Row 对象出错
        return [dict(row) for row in result.mappings().all()]

    async def create(self, user_data):
        """创建未签署合同模板"""
        db_templates = ContractStore(
            store_id=user_data.store_id,
            merchant_name=user_data.merchant_name,
            business_segment=user_data.business_segment,
            contact_phone=user_data.contact_phone,
            contract_url=user_data.contract_url,
            seal_url='0.0',
        )
        self.db.add(db_templates)
        await self.db.commit()
        await self.db.refresh(db_templates)
        return db_templates