#12

db/eloquent.pyi

@@ -0,0 +1,140 @@
+# pylint: disable = missing-class-docstring
+# pylint: disable = missing-function-docstring
+# pylint: disable = missing-module-docstring
+# pylint: disable = unused-argument
+
+from __future__ import annotations
+
+from typing import Any, Generic, Type, TypeVar, overload
+from typing_extensions import Self
+
+_ModelT = TypeVar ('_ModelT', bound = 'Model')
+
+
+class Connection:
+    def select (self, query: str, bindings: dict[str, Any] | None = None) -> Any: ...
+
+    def insert (self, query: str, bindings: dict[str, Any] | None = None) -> int: ...
+
+    def update (self, query: str, bindings: dict[str, Any] | None = None) -> int: ...
+
+    def delete (self, query: str, bindings: dict[str, Any] | None = None) -> int: ...
+
+    def transaction (self, callback: Any) -> Any: ...
+
+    def begin_transaction (self) -> None: ...
+
+    def commit (self) -> None: ...
+
+    def rollback (self) -> None: ...
+
+
+class ConnectionResolver:
+    def connection (self, name: str | None = None) -> Any: ...
+
+    def get_default_connection (self) -> str: ...
+
+    def set_default_connection (self, name: str) -> None: ...
+
+
+class DatabaseManager:
+    connections: dict[str, Connection]
+
+    def __init__ (self, config: dict[str, Any]) -> None: ...
+
+    def connection (self, name: str | None = None) -> Connection: ...
+
+    def disconnect (self, name: str | None = None) -> None: ...
+
+    def reconnect (self, name: str | None = None) -> Connection: ...
+
+    def get_connections (self) -> dict[str, Connection]: ...
+
+
+class Model:
+    id:             int
+
+    _Model__exists: bool
+
+    def has_one (
+            self,
+            related_model:  Type[_ModelT],
+            foreign_key:    str | None      = None,
+    ) -> _ModelT: ...
+
+    def has_many (
+            self,
+            related_model:  Type[_ModelT],
+            foreign_key:    str | None      = None,
+    ) -> list[_ModelT]: ...
+
+    def belongs_to (
+            self,
+            related_model:  Type[_ModelT],
+            foreign_key:    str | None      = None,
+    ) -> _ModelT: ...
+
+    def belongs_to_many (
+            self,
+            related_model:  Type[_ModelT],
+            foreign_key:    str | None      = None,
+    ) -> list[_ModelT]: ...
+
+    def save (self) -> None: ...
+
+    def delete (self) -> None: ...
+
+    @classmethod
+    def find (cls, id_: int) -> Self | None: ...
+
+    @classmethod
+    def query (
+            cls,
+    ) -> QueryBuilder[Self]: ...
+
+    @overload
+    @classmethod
+    def where (
+            cls,
+            field:      str,
+            operator:   str,
+            value:      Any,
+    ) -> QueryBuilder[Self]: ...
+
+    @overload
+    @classmethod
+    def where (cls, field: str, value: Any) -> QueryBuilder[Self]: ...
+
+    @classmethod
+    def where_not_in (
+            cls,
+            column: str,
+            values: list[Any] | tuple
+    ) -> QueryBuilder[Self]: ...
+
+    @classmethod
+    def where_not_null (cls, field: str) -> QueryBuilder[Self]: ...
+
+    @classmethod
+    def set_connection_resolver (cls, resolver: DatabaseManager) -> None: ...
+
+
+class QueryBuilder (Generic[_ModelT]):
+    def first (self) -> _ModelT | None: ...
+
+    def get (self) -> list[_ModelT]: ...
+
+    @overload
+    def where (
+            self,
+            field:      str,
+            operator:   str,
+            value:      Any,
+    ) -> QueryBuilder[_ModelT]: ...
+
+    @overload
+    def where (self, field: str, value: Any) -> QueryBuilder[_ModelT]: ...
+
+    def where_null (self, field: str) -> QueryBuilder[_ModelT]: ...
+
+    def _load_relation (self, relation_name: str) -> QueryBuilder[_ModelT]: ...

db/models.py

@@ -0,0 +1,155 @@
+# pylint: disable = missing-class-docstring
+# pylint: disable = missing-function-docstring
+
+"""
+ぼざクリ DB の構成
+"""
+
+from __future__ import annotations
+
+from datetime import date, datetime
+
+from db.my_eloquent import Model
+
+
+class Comment (Model):
+    # pylint: disable = too-many-instance-attributes
+
+    id:         int
+    video_id:   int
+    comment_no: int
+    user_id:    int
+    content:    str
+    posted_at:  datetime
+    nico_count: int
+    vpos_ms:    int
+
+    __timestamps__ = False
+
+    @property
+    def video (
+            self,
+    ) -> Video:
+        return self.belongs_to (Video)
+
+    @property
+    def user (
+            self,
+    ) -> User:
+        return self.belongs_to (User)
+
+    def upsert (
+            self,
+            *args:  str,
+    ) -> None:
+        super ().upsert ('video_id', 'comment_no')
+
+
+class Tag (Model):
+    id:     int
+    name:   str
+
+    __timestamps__ = False
+
+    @property
+    def video_tags (
+            self,
+    ) -> list[VideoTag]:
+        return self.has_many (VideoTag)
+
+
+class User (Model):
+    id:     int
+    code:   str
+
+    __timestamps__ = False
+
+    @property
+    def comments (
+            self,
+    ) -> list[Comment]:
+        return self.has_many (Comment)
+
+
+class Video (Model):
+    id:             int
+    code:           str
+    title:          str
+    description:    str
+    uploaded_at:    datetime
+    deleted_at:     datetime | None
+
+    __timestamps__ = False
+
+    @property
+    def video_histories (
+            self,
+    ) -> list[VideoHistory]:
+        return self.has_many (VideoHistory)
+
+    @property
+    def video_tags (
+            self,
+    ) -> list[VideoTag]:
+        return self.has_many (VideoTag)
+
+    @property
+    def comments (
+            self,
+    ) -> list[Comment]:
+        return self.has_many (Comment)
+
+    def upsert (
+            self,
+            *args:  str,
+    ) -> None:
+        super ().upsert ('code')
+
+
+class VideoHistory (Model):
+    id:             int
+    video_id:       int
+    fetched_at:     date
+    views_count:    int
+
+    __timestamps__ = False
+
+    @property
+    def video (
+            self,
+    ) -> Video:
+        return self.belongs_to (Video)
+
+    def upsert (
+            self,
+            *args:  str,
+    ) -> None:
+        super ().upsert ('video_id', 'fetched_at')
+
+
+class VideoTag (Model):
+    id:             int
+    video_id:       int
+    tag_id:         int
+    tagged_at:      date
+    untagged_at:    date | None
+
+    __timestamps__ = False
+
+    @property
+    def video (
+            self,
+    ) -> Video:
+        return self.belongs_to (Video)
+
+    @property
+    def tag (
+            self,
+    ) -> Tag:
+        return self.belongs_to (Tag)
+
+    def upsert (
+            self,
+            *args:  str,
+    ) -> None:
+        super ().upsert ('video_id', 'tag_id')

db/my_eloquent.py

@@ -0,0 +1,31 @@
+# pylint: disable = missing-class-docstring
+# pylint: disable = missing-function-docstring
+
+"""
+みてるぞ式魔改造（言ふほどか?）版 Eloquent
+"""
+
+import eloquent
+
+
+class DatabaseManager (eloquent.DatabaseManager):
+    pass
+
+
+class Model (eloquent.Model):
+    id: int
+
+    def upsert (
+            self,
+            *args:  str,
+    ) -> None:
+        q = self.query ()
+        for arg in args:
+            q = q.where (arg, getattr (self, arg))
+        row = q.first ()
+        if row is not None:
+            self.id = row.id
+            # pylint: disable = invalid-name
+            # pylint: disable = attribute-defined-outside-init
+            self._Model__exists = True
+        self.save ()