79 lines
2.4 KiB
Python
79 lines
2.4 KiB
Python
from datetime import datetime, timezone
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .auth import current_user
|
|
from .database import get_db
|
|
from .models import PhoneNumber, UserAccount
|
|
|
|
router = APIRouter(prefix="/api/v1/numbers", tags=["numbers"])
|
|
|
|
|
|
def serialize_phone_number(number: PhoneNumber) -> dict:
|
|
return {
|
|
"id": str(number.id),
|
|
"phone_number": number.phone_number,
|
|
"monthly_number_price": float(number.monthly_number_price),
|
|
"inbound_price": float(number.inbound_price),
|
|
"outbound_price": float(number.outbound_price),
|
|
"currency": number.currency,
|
|
"status": number.status,
|
|
"acquisition_type": number.acquisition_type,
|
|
"assigned_at": number.assigned_at,
|
|
"created_at": number.created_at,
|
|
}
|
|
|
|
|
|
@router.get("/catalog")
|
|
def list_catalog(
|
|
db: Session = Depends(get_db),
|
|
_: UserAccount = Depends(current_user),
|
|
) -> list[dict]:
|
|
numbers = db.scalars(
|
|
select(PhoneNumber)
|
|
.where(PhoneNumber.status == "available", PhoneNumber.user_id.is_(None))
|
|
.order_by(PhoneNumber.created_at.desc())
|
|
).all()
|
|
return [serialize_phone_number(number) for number in numbers]
|
|
|
|
|
|
@router.get("/mine")
|
|
def list_my_numbers(
|
|
db: Session = Depends(get_db),
|
|
user: UserAccount = Depends(current_user),
|
|
) -> list[dict]:
|
|
numbers = db.scalars(
|
|
select(PhoneNumber)
|
|
.where(PhoneNumber.user_id == user.id, PhoneNumber.status == "assigned")
|
|
.order_by(PhoneNumber.assigned_at.desc())
|
|
).all()
|
|
return [serialize_phone_number(number) for number in numbers]
|
|
|
|
|
|
@router.post("/{number_id}/purchase")
|
|
def purchase_number(
|
|
number_id: UUID,
|
|
db: Session = Depends(get_db),
|
|
user: UserAccount = Depends(current_user),
|
|
) -> dict:
|
|
number = db.scalar(
|
|
select(PhoneNumber).where(
|
|
PhoneNumber.id == number_id,
|
|
PhoneNumber.status == "available",
|
|
PhoneNumber.user_id.is_(None),
|
|
)
|
|
)
|
|
if not number:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Number is not available")
|
|
|
|
now = datetime.now(timezone.utc)
|
|
number.user_id = user.id
|
|
number.status = "assigned"
|
|
number.acquisition_type = "purchased"
|
|
number.assigned_at = now
|
|
db.commit()
|
|
db.refresh(number)
|
|
return serialize_phone_number(number)
|