Yves Gugger
9fde9eab4a
Some checks failed
CI / Frontend Lint & Type Check (push) Has been cancelled
CI / Frontend Build (push) Has been cancelled
CI / Backend Lint (push) Has been cancelled
CI / Backend Tests (push) Has been cancelled
CI / Docker Build (push) Has been cancelled
CI / Security Scan (push) Has been cancelled
Deploy / Build & Push Images (push) Has been cancelled
Deploy / Deploy to Server (push) Has been cancelled
Deploy / Notify (push) Has been cancelled
- Changed SubscriptionTier.STARTER to SCOUT (correct tier name) - Removed domain_limit from Subscription constructor (it's a computed property) - Database needs to be re-initialized with: python scripts/init_db.py && python scripts/seed_tld_prices.py
309 lines
8.8 KiB
Python
309 lines
8.8 KiB
Python
"""Domain management API (requires authentication)."""
|
|
from datetime import datetime
|
|
from math import ceil
|
|
|
|
from fastapi import APIRouter, HTTPException, status, Query
|
|
from sqlalchemy import select, func
|
|
|
|
from app.api.deps import Database, CurrentUser
|
|
from app.models.domain import Domain, DomainCheck, DomainStatus
|
|
from app.models.subscription import TIER_CONFIG, SubscriptionTier
|
|
from app.schemas.domain import DomainCreate, DomainResponse, DomainListResponse
|
|
from app.services.domain_checker import domain_checker
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("", response_model=DomainListResponse)
|
|
async def list_domains(
|
|
current_user: CurrentUser,
|
|
db: Database,
|
|
page: int = Query(1, ge=1),
|
|
per_page: int = Query(20, ge=1, le=100),
|
|
):
|
|
"""Get list of monitored domains for current user."""
|
|
# Count total
|
|
count_query = select(func.count(Domain.id)).where(Domain.user_id == current_user.id)
|
|
total = (await db.execute(count_query)).scalar()
|
|
|
|
# Get domains with pagination
|
|
offset = (page - 1) * per_page
|
|
query = (
|
|
select(Domain)
|
|
.where(Domain.user_id == current_user.id)
|
|
.order_by(Domain.created_at.desc())
|
|
.offset(offset)
|
|
.limit(per_page)
|
|
)
|
|
result = await db.execute(query)
|
|
domains = result.scalars().all()
|
|
|
|
return DomainListResponse(
|
|
domains=[DomainResponse.model_validate(d) for d in domains],
|
|
total=total,
|
|
page=page,
|
|
per_page=per_page,
|
|
pages=ceil(total / per_page) if total > 0 else 1,
|
|
)
|
|
|
|
|
|
@router.post("", response_model=DomainResponse, status_code=status.HTTP_201_CREATED)
|
|
async def add_domain(
|
|
domain_data: DomainCreate,
|
|
current_user: CurrentUser,
|
|
db: Database,
|
|
):
|
|
"""Add a domain to monitoring list."""
|
|
# Check subscription limit
|
|
await db.refresh(current_user, ["subscription", "domains"])
|
|
|
|
if current_user.subscription:
|
|
limit = current_user.subscription.max_domains
|
|
else:
|
|
limit = TIER_CONFIG[SubscriptionTier.SCOUT]["domain_limit"]
|
|
|
|
current_count = len(current_user.domains)
|
|
|
|
if current_count >= limit:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=f"Domain limit reached ({limit}). Upgrade your subscription to add more domains.",
|
|
)
|
|
|
|
# Check if domain already exists for this user
|
|
existing = await db.execute(
|
|
select(Domain).where(
|
|
Domain.user_id == current_user.id,
|
|
Domain.name == domain_data.name,
|
|
)
|
|
)
|
|
if existing.scalar_one_or_none():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Domain already in your monitoring list",
|
|
)
|
|
|
|
# Check domain availability
|
|
check_result = await domain_checker.check_domain(domain_data.name)
|
|
|
|
# Create domain
|
|
domain = Domain(
|
|
name=domain_data.name,
|
|
user_id=current_user.id,
|
|
status=check_result.status,
|
|
is_available=check_result.is_available,
|
|
registrar=check_result.registrar,
|
|
expiration_date=check_result.expiration_date,
|
|
notify_on_available=domain_data.notify_on_available,
|
|
last_checked=datetime.utcnow(),
|
|
)
|
|
db.add(domain)
|
|
await db.flush()
|
|
|
|
# Create initial check record
|
|
check = DomainCheck(
|
|
domain_id=domain.id,
|
|
status=check_result.status,
|
|
is_available=check_result.is_available,
|
|
response_data=str(check_result.to_dict()),
|
|
checked_at=datetime.utcnow(),
|
|
)
|
|
db.add(check)
|
|
|
|
await db.commit()
|
|
await db.refresh(domain)
|
|
|
|
return domain
|
|
|
|
|
|
@router.get("/{domain_id}", response_model=DomainResponse)
|
|
async def get_domain(
|
|
domain_id: int,
|
|
current_user: CurrentUser,
|
|
db: Database,
|
|
):
|
|
"""Get a specific domain."""
|
|
result = await db.execute(
|
|
select(Domain).where(
|
|
Domain.id == domain_id,
|
|
Domain.user_id == current_user.id,
|
|
)
|
|
)
|
|
domain = result.scalar_one_or_none()
|
|
|
|
if not domain:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Domain not found",
|
|
)
|
|
|
|
return domain
|
|
|
|
|
|
@router.delete("/{domain_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_domain(
|
|
domain_id: int,
|
|
current_user: CurrentUser,
|
|
db: Database,
|
|
):
|
|
"""Remove a domain from monitoring list."""
|
|
result = await db.execute(
|
|
select(Domain).where(
|
|
Domain.id == domain_id,
|
|
Domain.user_id == current_user.id,
|
|
)
|
|
)
|
|
domain = result.scalar_one_or_none()
|
|
|
|
if not domain:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Domain not found",
|
|
)
|
|
|
|
await db.delete(domain)
|
|
await db.commit()
|
|
|
|
|
|
@router.post("/{domain_id}/refresh", response_model=DomainResponse)
|
|
async def refresh_domain(
|
|
domain_id: int,
|
|
current_user: CurrentUser,
|
|
db: Database,
|
|
):
|
|
"""Manually refresh domain availability status."""
|
|
result = await db.execute(
|
|
select(Domain).where(
|
|
Domain.id == domain_id,
|
|
Domain.user_id == current_user.id,
|
|
)
|
|
)
|
|
domain = result.scalar_one_or_none()
|
|
|
|
if not domain:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Domain not found",
|
|
)
|
|
|
|
# Check domain
|
|
check_result = await domain_checker.check_domain(domain.name)
|
|
|
|
# Update domain
|
|
domain.status = check_result.status
|
|
domain.is_available = check_result.is_available
|
|
domain.registrar = check_result.registrar
|
|
domain.expiration_date = check_result.expiration_date
|
|
domain.last_checked = datetime.utcnow()
|
|
|
|
# Create check record
|
|
check = DomainCheck(
|
|
domain_id=domain.id,
|
|
status=check_result.status,
|
|
is_available=check_result.is_available,
|
|
response_data=str(check_result.to_dict()),
|
|
checked_at=datetime.utcnow(),
|
|
)
|
|
db.add(check)
|
|
|
|
await db.commit()
|
|
await db.refresh(domain)
|
|
|
|
return domain
|
|
|
|
|
|
@router.patch("/{domain_id}/notify", response_model=DomainResponse)
|
|
async def update_notification_settings(
|
|
domain_id: int,
|
|
notify_on_available: bool,
|
|
current_user: CurrentUser,
|
|
db: Database,
|
|
):
|
|
"""Update notification settings for a domain."""
|
|
result = await db.execute(
|
|
select(Domain).where(
|
|
Domain.id == domain_id,
|
|
Domain.user_id == current_user.id,
|
|
)
|
|
)
|
|
domain = result.scalar_one_or_none()
|
|
|
|
if not domain:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Domain not found",
|
|
)
|
|
|
|
domain.notify_on_available = notify_on_available
|
|
await db.commit()
|
|
await db.refresh(domain)
|
|
|
|
return domain
|
|
|
|
|
|
@router.get("/{domain_id}/history")
|
|
async def get_domain_history(
|
|
domain_id: int,
|
|
current_user: CurrentUser,
|
|
db: Database,
|
|
limit: int = Query(30, ge=1, le=365),
|
|
):
|
|
"""Get check history for a domain (Professional and Enterprise plans)."""
|
|
# Verify domain ownership
|
|
result = await db.execute(
|
|
select(Domain).where(
|
|
Domain.id == domain_id,
|
|
Domain.user_id == current_user.id,
|
|
)
|
|
)
|
|
domain = result.scalar_one_or_none()
|
|
|
|
if not domain:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Domain not found",
|
|
)
|
|
|
|
# Check subscription for history access
|
|
await db.refresh(current_user, ["subscription"])
|
|
if current_user.subscription:
|
|
history_days = current_user.subscription.history_days
|
|
if history_days == 0:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Check history requires Professional or Enterprise plan",
|
|
)
|
|
# Limit based on plan (-1 means unlimited)
|
|
if history_days > 0:
|
|
limit = min(limit, history_days)
|
|
else:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Check history requires Professional or Enterprise plan",
|
|
)
|
|
|
|
# Get check history
|
|
history_query = (
|
|
select(DomainCheck)
|
|
.where(DomainCheck.domain_id == domain_id)
|
|
.order_by(DomainCheck.checked_at.desc())
|
|
.limit(limit)
|
|
)
|
|
history_result = await db.execute(history_query)
|
|
checks = history_result.scalars().all()
|
|
|
|
return {
|
|
"domain": domain.name,
|
|
"total_checks": len(checks),
|
|
"history": [
|
|
{
|
|
"id": check.id,
|
|
"status": check.status.value if hasattr(check.status, 'value') else check.status,
|
|
"is_available": check.is_available,
|
|
"checked_at": check.checked_at.isoformat(),
|
|
}
|
|
for check in checks
|
|
]
|
|
}
|
|
|