Eloquent の型定義ファイルと型安全性確認；本番環境に取込み可能

eloquent.pyi

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+from typing import Any, Generic, Type, TypeVar, overload
+
+from eloquent.orm.relations.dynamic_property import DynamicProperty
+
+_TModel = TypeVar ('_TModel', 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
+
+    def has_one (
+            self,
+            related_model:  Type[Model],
+            foreign_key:    str | None  = None,
+    ) -> DynamicProperty: ...
+
+    def has_many (
+            self,
+            related_model:  Type[Model],
+            foreign_key:    str | None  = None,
+    ) -> DynamicProperty: ...
+
+    def belongs_to (
+            self,
+            related_model:  Type[Model],
+            foreign_key:    str | None  = None,
+    ) -> DynamicProperty: ...
+
+    def belongs_to_many (
+            self,
+            related_model:  Type[Model],
+            foreign_key:    str | None  = None,
+    ) -> DynamicProperty: ...
+
+    def save (self) -> None: ...
+
+    def delete (self) -> None: ...
+
+    @classmethod
+    def find (cls, id: int) -> Model | None: ...
+
+    @overload
+    @classmethod
+    def where (cls, field: str, operator: str, value: Any) -> QueryBuilder: ...
+
+    @overload
+    @classmethod
+    def where (cls, field: str, value: Any) -> QueryBuilder: ...
+
+    @classmethod
+    def where_not_in (cls, column: str, values: list[Any] | tuple) -> QueryBuilder: ...
+
+    @classmethod
+    def where_not_null (cls, field: str) -> QueryBuilder: ...
+
+    @classmethod
+    def set_connection_resolver (cls, resolver: DatabaseManager) -> None: ...
+
+
+class QueryBuilder (Generic[_TModel]):
+    def first (self) -> _TModel | None: ...
+
+    def get (self) -> list[_TModel]: ...
+
+    @overload
+    def where (self, field: str, operator: str, value: Any) -> QueryBuilder: ...
+
+    @overload
+    def where (self, field: str, value: Any) -> QueryBuilder: ...
+
+    def where_null (self, field: str) -> QueryBuilder: ...

eloquent/orm/relations/dynamic_property.pyi

@@ -0,0 +1,4 @@
+from eloquent import Model
+
+
+class DynamicProperty (Model): ...

update_db.py

@@ -204,6 +204,14 @@ def search_nico_by_tags (
 class Comment (Model):
     __timestamps__ = False
 
+    video_id:   int
+    comment_no: int
+    user_id:    int
+    content:    str
+    posted_at:  datetime
+    nico_count: int
+    vpos_ms:    int
+
     @property
     def video (
             self,
@@ -216,10 +224,22 @@ class Comment (Model):
     ) -> DynamicProperty:
         return self.belongs_to (User)
 
+    def upsert (
+            self,
+    ) -> None:
+        row = (Comment.where ('video_id', self.video_id)
+                      .where ('comment_no', self.comment_no)
+                      .first ())
+        if row is not None:
+            self.id = row.id
+        self.save ()
+
 
 class Tag (Model):
     __timestamps__ = False
 
+    name:   str
+
     @property
     def video_tags (
             self,
@@ -230,6 +250,8 @@ class Tag (Model):
 class User (Model):
     __timestamps__ = False
 
+    code:   str
+
     @property
     def comments (
             self,
@@ -240,6 +262,12 @@ class User (Model):
 class Video (Model):
     __timestamps__ = False
 
+    code:           str
+    title:          str
+    description:    str
+    uploaded_at:    datetime
+    deleted_at:     datetime | None
+
     @property
     def video_histories (
             self,
@@ -270,6 +298,10 @@ class Video (Model):
 class VideoHistory (Model):
     __timestamps__ = False
 
+    video_id:       int
+    fetched_at:     date
+    views_count:    int
+
     @property
     def video (
             self,
@@ -279,8 +311,7 @@ class VideoHistory (Model):
     def upsert (
             self,
     ) -> None:
-        row = (Video
+        row = (VideoHistory.where ('video_id', self.video_id)
-                .where ('video_id', self.video_id)
                            .where ('fetched_at', self.fetched_at)
                            .first ())
         if row is not None:
@@ -291,6 +322,11 @@ class VideoHistory (Model):
 class VideoTag (Model):
     __timestamps__ = False
 
+    video_id:       int
+    tag_id:         int
+    tagged_at:      date
+    untagged_at:    date | None
+
     @property
     def video (
             self,
@@ -306,8 +342,7 @@ class VideoTag (Model):
     def upsert (
             self,
     ) -> None:
-        row = (Video
+        row = (VideoTag.where ('video_id', self.video_id)
-                .where ('video_id', self.video_id)
                        .where ('tag_id', self.tag_id)
                        .first ())
         if row is not None:
@@ -358,53 +393,8 @@ class CommentResult (TypedDict):
     isMyPost:       bool
 
 
-class CommentRow (TypedDict):
-    id:         int
-    video_id:   int
-    comment_no: int
-    user_id:    int
-    content:    str
-    posted_at:  datetime
-    nico_count: int
-    vpos_ms:    int | None
-
-
-class TagRow (TypedDict):
-    id:     int
-    name:   str
-
-
-class UserRow (TypedDict):
-    id:     int
-    code:   str
-
-
-class VideoRow (TypedDict):
-    id:             int
-    code:           str
-    title:          str
-    description:    str
-    uploaded_at:    datetime
-    deleted_at:     datetime | None
-
-
-class VideoHistoryRow (TypedDict):
-    id:             int
-    video_id:       int
-    fetched_at:     date
-    views_count:    int
-
-
-class VideoTagRow (TypedDict):
-    id:             int
-    video_id:       int
-    tag_id:         int
-    tagged_at:      date
-    untagged_at:    date | None
-
-
 def normalise (
-        s:  str
+        s:  str,
 ) -> str:
     return unicodedata.normalize ('NFKC', s).lower ()