51 lines
1.3 KiB
Python
51 lines
1.3 KiB
Python
from fastapi import Depends, HTTPException, status
|
|
from typing import Annotated
|
|
from sqlalchemy.orm import Session
|
|
from uuid import UUID
|
|
|
|
from ...dependencies import get_db
|
|
from ...db import models
|
|
|
|
from ..auth import services as auth_services
|
|
from ..auth import schemas as auth_schemas
|
|
|
|
from . import schemas
|
|
|
|
|
|
def get_news(
|
|
db: Annotated[Session, Depends(get_db)],
|
|
) -> list[schemas.NewsInDb]:
|
|
return [
|
|
schemas.NewsInDb.model_validate(n)
|
|
for n in db.query(models.News).order_by(models.News.created.desc()).all()
|
|
]
|
|
|
|
|
|
def create_news(
|
|
news: schemas.CreateNews,
|
|
current_user: auth_schemas.UserInDB,
|
|
db: Session,
|
|
) -> schemas.NewsInDb:
|
|
if current_user.username == "admin":
|
|
n = models.News(title=news.title, content=news.content)
|
|
db.add(n)
|
|
db.commit()
|
|
return schemas.NewsInDb.model_validate(n)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Could not validate credentials",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
|
|
def tap_news(news_id: UUID, db: Session):
|
|
n = db.query(models.News).filter(models.News.id == news_id).first()
|
|
if n:
|
|
setattr(n, "taps", n.taps + 1)
|
|
db.commit()
|
|
return
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Not found",
|
|
)
|