Crud test

app/containers.py

@@ -6,7 +6,7 @@
 
 class AppContainer(containers.DeclarativeContainer):
     wiring_config = containers.WiringConfiguration(
-        packages=(".features.content",)
+        packages=(".features.content", ".features.twitch")
     )
     config = providers.Configuration()
     resources = providers.Container(

app/features/__init__.py

@@ -1,6 +1,7 @@
 from dependency_injector import containers, providers
 
 from .content.containers import ContentContainer
+from .twitch.containers import TwitchContainer
 
 
 class FeaturesContainer(containers.DeclarativeContainer):
@@ -9,3 +10,6 @@ class FeaturesContainer(containers.DeclarativeContainer):
     content = providers.Container(
         ContentContainer, resources=resources, config=config.content
     )
+    twitch = providers.Container(
+        TwitchContainer, resources=resources, config=config.bot
+    )

app/features/admin/__init__.py

@@ -4,9 +4,9 @@
 
 # /admin/ depends on token
 # /admin/login POST -> redis as store backend
-# /admin/logout POST -> redis as tore backend
+# /admin/logout POST -> redis as store backend
 # /admin/<model-name>/ GET -> list of items with pagination
 # -> GET /:id -> get item for edit
 # -> PATCH /:id -> update item if exists
 # -> POST /:id -> create item
-# -> DELETE /:id -> delete item
+# -> DELETE /:id -> delete item

app/features/content/models.py

@@ -1,5 +1,6 @@
 import typing as T
 
+from datetime import datetime
 from pydantic import BaseModel
 
 

app/features/twitch/__init__.py

@@ -0,0 +1,3 @@
+from fastapi import APIRouter
+
+API = APIRouter()

app/features/twitch/api.py

@@ -0,0 +1,48 @@
+from fastapi import Depends
+from dependency_injector.wiring import inject, Provide
+
+from ...resources.services import ModelDataService
+from ...containers import AppContainer
+
+from . import API
+from .models import Bot, BotIn, Channel, ChannelIn
+
+CONTAINER = AppContainer.features.twitch
+
+
+@API.get("/bots", response_model=list[Bot])
+@inject
+async def get_bot_many(
+    query=Depends(Provide[CONTAINER.bot.query])
+):
+    return await query.all()
+
+
+@API.post("/bot", response_model=Bot)
+@inject
+async def create_bot(
+    bot_in: BotIn,
+    query: ModelDataService = Depends(Provide[CONTAINER.bot.query])
+):
+    bot: Bot = await query.create(bot_in)
+    await query.commit()
+    return bot
+
+
+@API.get("/channels", response_model=list[Channel])
+@inject
+async def get_channel_many(
+    query=Depends(Provide[CONTAINER.channel.query])
+):
+    return await query.all()
+
+
+@API.post("/channel", response_model=Channel)
+@inject
+async def create_channel(
+    channel_in: ChannelIn,
+    query: ModelDataService = Depends(Provide[CONTAINER.channel.query])
+):
+    channel: Channel = await query.create(channel_in)
+    await query.commit()
+    return channel

app/features/twitch/containers.py

@@ -0,0 +1,28 @@
+from dependency_injector import containers, providers
+
+from ...resources.containers import ModelDataContainer
+
+from . import API
+from .models import Bot, Channel
+from .tables import BOT_TABLE, CHANNEL_TABLE
+
+
+class TwitchContainer(containers.DeclarativeContainer):
+    config = providers.Configuration()
+    resources = providers.DependenciesContainer(
+        db=providers.DependenciesContainer()
+    )
+
+    api = providers.Object(API)
+    bot = providers.Container(
+        ModelDataContainer,
+        db=resources.db,
+        model=Bot,
+        table=BOT_TABLE,
+    )
+    channel = providers.Container(
+        ModelDataContainer,
+        db=resources.db,
+        model=Channel,
+        table=CHANNEL_TABLE
+    )

app/features/twitch/models.py

@@ -0,0 +1,56 @@
+import typing as T
+
+from datetime import datetime
+from pydantic import BaseModel
+
+
+class Bot(BaseModel):
+    id: int
+    secret_id: str
+    client_id: str
+    nickname: str
+    load: int = 0
+    max_load: int = 40
+
+
+class BotIn(BaseModel):
+    secret_id: str
+    client_id: str
+    nickname: str
+
+
+class User(BaseModel):
+    id: int
+    nicknames: list[str]
+    ext_id: int
+
+
+class Message(BaseModel):
+    id: int
+    content: str
+    user_id: int
+    channel_id: int
+
+
+class ChannelIn(BaseModel):
+    url: str
+
+
+class Channel(ChannelIn):
+    id: int
+    url: str
+    start_listen_at: T.Optional[datetime]
+    ping_listen_at: T.Optional[datetime]
+    stop_listen_at: T.Optional[datetime]
+    listen_bot_id: T.Optional[int]
+    ignore: bool = False
+    active: bool = False
+
+
+class ChannelAcquire(BaseModel):
+    size: int
+
+
+class BotWithChannels(BaseModel):
+    bot: Bot
+    channels: list[Channel]

app/features/twitch/tables.py

@@ -0,0 +1,71 @@
+import sqlalchemy as sa
+
+from ...resources.providers import METADATA
+
+BOT_TABLE = sa.Table(
+    "bot",
+    METADATA,
+    sa.Column("id", sa.Integer, primary_key=True, nullable=False),
+    sa.Column("secret_id", sa.String(255), nullable=False),
+    sa.Column("client_id", sa.String(255), nullable=False),
+    sa.Column("nickname", sa.String(255), nullable=False),
+    sa.Column(
+        "load",
+        sa.Integer,
+        nullable=False,
+        default=0,
+    ),
+    sa.Column(
+        "max_load",
+        sa.Integer,
+        nullable=False,
+        default=40,
+    ),
+)
+
+USER_TABLE = sa.Table(
+    "user",
+    METADATA,
+    sa.Column("id", sa.Integer, primary_key=True, nullable=False),
+    sa.Column("nicknames", sa.ARRAY(sa.String(255)), nullable=False),
+    sa.Column("ext_id", sa.Integer, nullable=False),
+)
+
+MESSAGE_TABLE = sa.Table(
+    "message",
+    METADATA,
+    sa.Column("id", sa.Integer, primary_key=True, nullable=False),
+    sa.Column("content", sa.String(255), nullable=False),
+    sa.Column("user_id", sa.Integer, sa.ForeignKey("user.id"), nullable=False),
+    sa.Column(
+        "channel_id",
+        sa.Integer,
+        sa.ForeignKey("channel.id"),
+        nullable=False,
+    ),
+)
+
+CHANNEL_TABLE = sa.Table(
+    "channel",
+    METADATA,
+    sa.Column("id", sa.Integer, primary_key=True, nullable=False),
+    sa.Column("url", sa.String(255), nullable=False),
+    sa.Column("start_listen_at", sa.DateTime(timezone=True), nullable=True),
+    sa.Column("ping_listen_at", sa.DateTime(timezone=True), nullable=True),
+    sa.Column("stop_listen_at", sa.DateTime(timezone=True), nullable=True),
+    sa.Column(
+        "listen_bot_id", sa.Integer, sa.ForeignKey("bot.id"), nullable=True
+    ),
+    sa.Column(
+        "ignore",
+        sa.Boolean,
+        nullable=False,
+        default=False,
+    ),
+    sa.Column(
+        "active",
+        sa.Boolean,
+        nullable=False,
+        default=False,
+    ),
+)

app/features/twitch/tasks.py

@@ -0,0 +1,32 @@
+import asyncio
+
+import httpx
+
+from celery import shared_task
+from celery.utils.log import get_task_logger
+from dependency_injector.wiring import Provide, inject
+
+from ...resources.services import ModelDataService
+from ...containers import AppContainer
+
+logger = get_task_logger(__name__)
+
+
+@inject
+async def fetch_content_async(
+    content_id,
+    query: ModelDataService = Provide[
+        AppContainer.features.content.data.query
+    ],
+):
+    content = await query.get(content_id)
+    content.body = httpx.get(content.url).text
+    await query.update(content)
+    await query.commit()
+    logger.info([content.id, len(content.body)])
+
+
+@shared_task(autoretry_for=(Exception,), name="fetch_content")
+def fetch_content(content_id: int):
+    loop = asyncio.get_event_loop()
+    loop.run_until_complete(fetch_content_async(content_id))

app/resources/services.py

@@ -67,7 +67,7 @@ async def filter(
             query = query.limit(limit)
         if offset:
             query = query.offset(offset)
-        if order_by is None:
+        if order_by is not None:
             query = query.order_by(order_by)
         return (
             await self.cursor.one(query)
@@ -91,5 +91,7 @@ async def update(self, model, exclude: tuple = ("id",), query=None):
         )
 
     async def delete(self, id_: int = None, query=None):
-        query = query or (self.table.c.id == id_)
-        await self.cursor.one(self.table.delete(query))
+        query = query or sa.delete(self.table)\
+                .where(self.table.c.id == id_)\
+                .returning(self.table)
+        return await self.cursor.one(query)

migrations/versions/0001_initial_migration.py

@@ -19,17 +19,65 @@
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
     op.create_table(
-        "content",
+        "bot",
         sa.Column("id", sa.Integer(), nullable=False),
-        sa.Column("url", sa.String(length=1024), nullable=False),
-        sa.Column("body", sa.Text(), nullable=True),
+        sa.Column("secret_id", sa.String(length=255), nullable=False),
+        sa.Column("client_id", sa.String(length=255), nullable=False),
+        sa.Column("nickname", sa.String(length=255), nullable=False),
+        sa.Column("load", sa.Integer(), nullable=False),
+        sa.Column("max_load", sa.Integer(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_table(
+        "user",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column(
+            "nicknames", sa.ARRAY(sa.String(length=255)), nullable=False
+        ),
+        sa.Column("ext_id", sa.Integer(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_table(
+        "channel",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("url", sa.String(length=255), nullable=False),
+        sa.Column(
+            "start_listen_at", sa.DateTime(timezone=True), nullable=True
+        ),
+        sa.Column("ping_listen_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("stop_listen_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("listen_bot_id", sa.Integer(), nullable=True),
+        sa.Column("ignore", sa.Boolean(), nullable=False),
+        sa.Column("active", sa.Boolean(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["listen_bot_id"],
+            ["bot.id"],
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_table(
+        "message",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("content", sa.String(length=255), nullable=False),
+        sa.Column("user_id", sa.Integer(), nullable=False),
+        sa.Column("channel_id", sa.Integer(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["channel_id"],
+            ["channel.id"],
+        ),
+        sa.ForeignKeyConstraint(
+            ["user_id"],
+            ["user.id"],
+        ),
         sa.PrimaryKeyConstraint("id"),
-        sa.UniqueConstraint("url"),
     )
     # ### end Alembic commands ###
 
 
 def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.drop_table("content")
-    # ### end Alembic commands ###
+    op.drop_table("message")
+    op.drop_table("channel")
+    op.drop_table("user")
+    op.drop_table("bot")
+    # ### end Alembic commands ###