59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
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) |