didier/database/crud/bookmarks.py

74 lines
2.5 KiB
Python
Raw Normal View History

2022-08-30 01:32:46 +02:00
from typing import Optional
import sqlalchemy.exc
2022-09-01 01:02:18 +02:00
from sqlalchemy import delete, func, select
2022-08-30 01:32:46 +02:00
from sqlalchemy.ext.asyncio import AsyncSession
from database.crud.users import get_or_add_user
2022-09-01 01:02:18 +02:00
from database.exceptions import (
DuplicateInsertException,
Forbidden,
ForbiddenNameException,
NoResultFoundException,
)
2022-08-30 01:32:46 +02:00
from database.schemas import Bookmark
__all__ = ["create_bookmark", "get_bookmarks", "get_bookmark_by_name"]
async def create_bookmark(session: AsyncSession, user_id: int, label: str, jump_url: str) -> Bookmark:
"""Create a new bookmark to a message"""
# Don't allow bookmarks with names of subcommands
2022-09-01 01:02:18 +02:00
if label.lower() in ["create", "delete", "ls", "list", "Rm", "search"]:
2022-08-30 01:32:46 +02:00
raise ForbiddenNameException
await get_or_add_user(session, user_id)
try:
bookmark = Bookmark(label=label, jump_url=jump_url, user_id=user_id)
session.add(bookmark)
await session.commit()
await session.refresh(bookmark)
except sqlalchemy.exc.IntegrityError as e:
raise DuplicateInsertException from e
return bookmark
2022-09-01 01:02:18 +02:00
async def delete_bookmark_by_id(session: AsyncSession, user_id: int, bookmark_id: int):
"""Find a bookmark by its id & delete it
This fails if you don't own this bookmark
"""
2022-09-01 01:26:46 +02:00
select_statement = select(Bookmark).where(Bookmark.bookmark_id == bookmark_id)
bookmark = (await session.execute(select_statement)).scalar_one_or_none()
2022-09-01 01:02:18 +02:00
# No bookmark with this id
if bookmark is None:
raise NoResultFoundException
# You don't own this bookmark
if bookmark.user_id != user_id:
raise Forbidden
# Delete it
2022-09-01 01:26:46 +02:00
delete_statement = delete(Bookmark).where(Bookmark.bookmark_id == bookmark_id)
await session.execute(delete_statement)
2022-09-01 01:02:18 +02:00
await session.commit()
2022-08-30 01:32:46 +02:00
async def get_bookmarks(session: AsyncSession, user_id: int, *, query: Optional[str] = None) -> list[Bookmark]:
"""Get all a user's bookmarks"""
statement = select(Bookmark).where(Bookmark.user_id == user_id)
if query is not None:
statement = statement.where(Bookmark.label.ilike(f"%{query.lower()}%"))
2023-07-08 01:23:47 +02:00
return list((await session.execute(statement)).scalars().all())
2022-08-30 01:32:46 +02:00
async def get_bookmark_by_name(session: AsyncSession, user_id: int, query: str) -> Optional[Bookmark]:
"""Try to find a bookmark by its name"""
statement = select(Bookmark).where(Bookmark.user_id == user_id).where(func.lower(Bookmark.label) == query.lower())
return (await session.execute(statement)).scalar_one_or_none()