from typing import List, Optional from uuid import UUID from sqlalchemy.orm import Session from models.shortened_url import ShortenedUrl from schemas.shortened_url import ShortenedUrlSchema, ShortenedUrlCreate from fastapi import HTTPException def get_user_shortened_urls(db: Session, user_id: UUID, skip: int = 0, limit: int = 100) -> List[ShortenedUrlSchema]: """ Retrieves a paginated list of shortened URLs for a given user. Args: db (Session): The database session. user_id (UUID): The ID of the user. skip (int): The number of items to skip (for pagination). limit (int): The maximum number of items to return. Returns: List[ShortenedUrlSchema]: A list of ShortenedUrlSchema objects. """ urls = db.query(ShortenedUrl).filter(ShortenedUrl.user_id == user_id).offset(skip).limit(limit).all() return [ShortenedUrlSchema.from_orm(url) for url in urls] def get_shortened_url_by_id(db: Session, url_id: UUID) -> Optional[ShortenedUrlSchema]: """ Retrieves a shortened URL by its ID. Args: db (Session): The database session. url_id (UUID): The ID of the shortened URL. Returns: Optional[ShortenedUrlSchema]: The ShortenedUrlSchema object if found, otherwise None. """ url = db.query(ShortenedUrl).filter(ShortenedUrl.id == url_id).first() if not url: return None return ShortenedUrlSchema.from_orm(url) def create_shortened_url(db: Session, url_data: ShortenedUrlCreate) -> ShortenedUrlSchema: """ Creates a new shortened URL in the database. Args: db (Session): The database session. url_data (ShortenedUrlCreate): The data for the shortened URL to create. Returns: ShortenedUrlSchema: The newly created ShortenedUrlSchema object. """ existing_url = db.query(ShortenedUrl).filter(ShortenedUrl.shortened_url == url_data.shortened_url).first() if existing_url: raise HTTPException(status_code=400, detail="Shortened URL already exists") db_url = ShortenedUrl(**url_data.dict()) db.add(db_url) db.commit() db.refresh(db_url) return ShortenedUrlSchema.from_orm(db_url)