93 lines
2.7 KiB
Python
93 lines
2.7 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
from typing import List
|
|
|
|
from app.models import ChannelDB, ChannelCreate, ChannelResponse
|
|
from app.utils.database import get_db
|
|
from app.auth.dependencies import get_current_user, require_roles
|
|
from app.models.auth import CognitoUser
|
|
|
|
router = APIRouter(
|
|
prefix="/channels",
|
|
tags=["channels"]
|
|
)
|
|
|
|
@router.post("/", response_model=ChannelResponse, status_code=status.HTTP_201_CREATED)
|
|
@require_roles("admin")
|
|
def create_channel(
|
|
channel: ChannelCreate,
|
|
db: Session = Depends(get_db),
|
|
user: CognitoUser = Depends(get_current_user)
|
|
):
|
|
"""Create a new channel"""
|
|
db_channel = ChannelDB(**channel.model_dump())
|
|
db.add(db_channel)
|
|
db.commit()
|
|
db.refresh(db_channel)
|
|
return db_channel
|
|
|
|
@router.get("/{tvg_id}", response_model=ChannelResponse)
|
|
def get_channel(
|
|
tvg_id: str,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Get a channel by tvg_id"""
|
|
channel = db.query(ChannelDB).filter(ChannelDB.tvg_id == tvg_id).first()
|
|
if not channel:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Channel not found"
|
|
)
|
|
return channel
|
|
|
|
@router.put("/{tvg_id}", response_model=ChannelResponse)
|
|
@require_roles("admin")
|
|
def update_channel(
|
|
tvg_id: str,
|
|
channel: ChannelCreate,
|
|
db: Session = Depends(get_db),
|
|
user: CognitoUser = Depends(get_current_user)
|
|
):
|
|
"""Update a channel"""
|
|
db_channel = db.query(ChannelDB).filter(ChannelDB.tvg_id == tvg_id).first()
|
|
if not db_channel:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Channel not found"
|
|
)
|
|
|
|
for key, value in channel.model_dump().items():
|
|
setattr(db_channel, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_channel)
|
|
return db_channel
|
|
|
|
@router.delete("/{tvg_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
@require_roles("admin")
|
|
def delete_channel(
|
|
tvg_id: str,
|
|
db: Session = Depends(get_db),
|
|
user: CognitoUser = Depends(get_current_user)
|
|
):
|
|
"""Delete a channel"""
|
|
channel = db.query(ChannelDB).filter(ChannelDB.tvg_id == tvg_id).first()
|
|
if not channel:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Channel not found"
|
|
)
|
|
db.delete(channel)
|
|
db.commit()
|
|
return None
|
|
|
|
@router.get("/", response_model=List[ChannelResponse])
|
|
@require_roles("admin")
|
|
def list_channels(
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
db: Session = Depends(get_db),
|
|
user: CognitoUser = Depends(get_current_user)
|
|
):
|
|
"""List all channels with pagination"""
|
|
return db.query(ChannelDB).offset(skip).limit(limit).all() |