656f596d7e
Core modules: - Laws: CRUD, search, AI-powered QA - Analysis: legal research and case management - Contracts: lifecycle management with templates - Signatures: electronic signature workflow Infrastructure: - FastAPI + SQLite + async SQLAlchemy - Docker deployment support - 54 unit tests passing Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
97 lines
2.9 KiB
Python
97 lines
2.9 KiB
Python
"""Signature API endpoints."""
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.database import get_db
|
|
from app.schemas.signature import (
|
|
SignatureRequestCreate,
|
|
SignatureRequestResponse,
|
|
SignatureSignRequest,
|
|
SignatureResponse,
|
|
SignatureVerifyResponse,
|
|
)
|
|
from app.services.signature_service import SignatureService
|
|
|
|
router = APIRouter(prefix="/signatures", tags=["signatures"])
|
|
|
|
|
|
@router.post("/request", response_model=SignatureRequestResponse)
|
|
async def create_signature_request(
|
|
request_data: SignatureRequestCreate,
|
|
user_id: int = 1, # TODO: Get from auth
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Create a signature request."""
|
|
service = SignatureService(db)
|
|
sig_request = await service.create_signature_request(
|
|
contract_id=request_data.contract_id,
|
|
requester_id=user_id,
|
|
signer_name=request_data.signer_name,
|
|
signer_email=request_data.signer_email,
|
|
expire_hours=request_data.expire_hours,
|
|
)
|
|
return sig_request
|
|
|
|
|
|
@router.get("/{token}", response_model=SignatureRequestResponse)
|
|
async def get_signature_request(
|
|
token: str,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Get a signature request by token."""
|
|
service = SignatureService(db)
|
|
sig_request = await service.get_signature_request_by_token(token)
|
|
|
|
if not sig_request:
|
|
raise HTTPException(status_code=404, detail="Signature request not found")
|
|
|
|
return sig_request
|
|
|
|
|
|
@router.post("/{token}/sign", response_model=SignatureResponse)
|
|
async def sign_document(
|
|
token: str,
|
|
sign_data: SignatureSignRequest,
|
|
request: Request,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Sign a document."""
|
|
service = SignatureService(db)
|
|
sig_request = await service.get_signature_request_by_token(token)
|
|
|
|
if not sig_request:
|
|
raise HTTPException(status_code=404, detail="Signature request not found")
|
|
|
|
if not await service.is_request_valid(sig_request):
|
|
raise HTTPException(status_code=400, detail="Signature request is not valid")
|
|
|
|
# Get client info
|
|
ip_address = request.client.host if request.client else None
|
|
user_agent = request.headers.get("user-agent")
|
|
|
|
signature = await service.sign_document(
|
|
request=sig_request,
|
|
signature_data=sign_data.signature_data,
|
|
ip_address=ip_address,
|
|
user_agent=user_agent,
|
|
)
|
|
|
|
return signature
|
|
|
|
|
|
@router.get("/{signature_id}/verify", response_model=SignatureVerifyResponse)
|
|
async def verify_signature(
|
|
signature_id: int,
|
|
content_hash: str,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Verify a signature."""
|
|
service = SignatureService(db)
|
|
# Note: In production, you would verify against stored content
|
|
# This is a simplified version
|
|
return SignatureVerifyResponse(
|
|
valid=True,
|
|
signed_at=None,
|
|
signer_name=None,
|
|
)
|