usrsem / kaif-telegram-service Goto Github PK

Task

Work in dtos.py module.

Create FileType enum:

• MP4
• JPEG

Create Story dataclass with following fields:

• event_id: int
• client_id: int
• story_id: int
• media_type: FileType
• media_path: str
• tags: str

Problem explanation

We need this service to call business logic from handlers

Task

Declare StoriesService protocol

• init_story(event_id: int, client_id: int) -> None
• add_media(absolute_path_to_media: str, file_type: FileType, client_id: int) -> None
• add_tags(tags: str, client_id: int) -> None
• save_story(client_id: itn) -> None

Implementation's __init__ method should get StoriesCache, StoriesPushService and KaifCrudService

Problem explanation

While clients will be sending their stories, service must save parts of their requests. Like media first and then tags. There is a need to use cache with time to live feature, because client can forgot to finish sending a story.

Task

Create generic protocol with get, add and __contains__ . Generic types is K for keys and V for values.

Implementation should use TTLCache from cachetools package. __init__ gets time to live in minutes, default is 60.

Task

Add factory to factories.py module

Problem explanation

While working with stories, there will be a need to send messages to client, therefore we need a separated service for that.

Task

Protocol

Create StoriesPushService protocol with following methods:

• send_media_request_message() -> None text="Send media (photo/video/rounded video)"
• send_media_error_message() -> None text="Error while loading media, try again"
• send_tags_request_message() -> None text="Send tags for story", reply_mrkup=no_tags)
• send_tags_error_message() -> None text="Error while parsing tags, try again"
• send_done_message() -> None text="Story saved !"

Implementation

Class should work with Bot object by using aggregation, also it should send logs with log object from loader.py module

Problem explanation

While working with bot it's cool and handy to use keyboards!

Task

Create module stories.py in keyboards package. It chould contain the following factories:

• get_empty_tags_keyboard() -> InlineKeyboardMarkup should return one button with text "Send without tags" and data "stories:no_tags"
• get_send_story_keyboard(event_id: int) -> InlineKeyboardMarkup should return one button with text "Send new story" and data f"stories:new_{event_id}"

Tips

Create constructor for keyboards in separate module.

def create_keyborad(
    text_data: tuple[tuple[str, str], ...],
    row_width: int = 1
) -> InlineKeyboardMarkup:
    keyboard = InlineKeyboardMarkup(row_width)

    row_btns = (
        InlineKeyboardButton(text, callback_data=data)
        for text, data in text_data
    )

    keyboard.add(*row_btns)
    return keyboard
    

Use it like this:

text_data = (("Send new story", "stories:new"),)
keyboard = create_keyborad(text_data)

Task

Create object of StoriesCache protocol in loader.py module.

Problem explanation

Handlers is used to process client's requests, run business logic from service layer and produce responses.

Task

Add stories.py module to handlers package. This file should implement the following functions:

• new_story(query: CallbackQuery) -> None should catch query with data "stories:new_{event_id}", parse event_id from data and invoke init_story method of StoriesService.
• add_media(query: CallbackQuery) -> None should work with all types of FileType enum. Save media in ./media/stories/{event_id} folder in project root directory. Pattern for file name f"{client_id}.{ext}".
• add_no_tags(query: CallbackQuery) -> None should parse catch query with data "stories:no_data".
• add_tags(message: Message) -> None should catch text.

Note, that each handler can create StoriesService with factory. Don't forget to add import of dp in __init__.py module in handlers package.

Think about problem that may arise in the future with add_tags method and how solve it.

Tips

Write lambda for catching only query with stories prefix. Examples could be found in idevtier_bot.

Problem explanation

kaif-telegram-service works with multiple models such as Client, Event and Story. There is a need to create single CRUD protocol for doing this stuff.

Task

KaifCrudService protocol should have the following structure:

• create(record: T) -> None
• read(id: K) -> T
• update(record: T) -> None
• delete(id: K) -> None

Where T and K are generic types for DTO and id. No need to write an implementation for production, just create fake class for testing in conftest.py. Note, that read method returns T, not an Optional[T].

Simple Generic Protocol implementation:

from typing import Generic, Protocol, TypeVar


T = TypeVar("T", contravariant=True)


class Foo(Generic[T], Protocol):
    def do(self, a: T) -> None: ...


class Bar:
    def do(self, a: int) -> None:
        print(a)


baz: Foo = Bar()