もろもろのモロヘイヤ

eloquent.pyi

@@ -51,6 +51,10 @@ class DatabaseManager:
 
 
 class Model:
+    id:             int
+
+    _Model__exists: bool
+
     def has_one (
             self,
             related_model:  Type[_ModelT],
@@ -82,6 +86,11 @@ class Model:
     @classmethod
     def find (cls, id_: int) -> Self | None: ...
 
+    @classmethod
+    def query (
+            cls,
+    ) -> QueryBuilder[Self]: ...
+
     @overload
     @classmethod
     def where (

models.py

@@ -0,0 +1,169 @@
+# pylint: disable = missing-class-docstring
+# pylint: disable = missing-function-docstring
+
+"""
+ぼざクリ DB の構成
+"""
+
+from __future__ import annotations
+
+from datetime import date, datetime
+
+from 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,
+    ) -> None:
+        upsert (self, 'video_id', 'comment_no')
+
+
+class Tag (Model):
+    id:     int
+    name:   str
+
+    __timestamps__ = False
+
+    @property
+    def video_tags (
+            self,
+    ) -> VideoTag:
+        return self.has_many (VideoTag)
+
+
+class User (Model):
+    id:     int
+    code:   str
+
+    __timestamps__ = False
+
+    @property
+    def comments (
+            self,
+    ) -> 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,
+    ) -> VideoHistory:
+        return self.has_many (VideoHistory)
+
+    @property
+    def video_tags (
+            self,
+    ) -> VideoTag:
+        return self.has_many (VideoTag)
+
+    @property
+    def comments (
+            self,
+    ) -> Comment:
+        return self.has_many (Comment)
+
+    def upsert (
+            self,
+    ) -> None:
+        upsert (self, '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,
+    ) -> None:
+        upsert (self, '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,
+    ) -> None:
+        upsert (self, 'video_id', 'tag_id')
+
+
+def upsert (
+        model:  Model,
+        *args:  str,
+) -> None:
+    q = model.query ()
+
+    for arg in args:
+        q = q.where (arg, getattr (model, arg))
+
+    row = q.first ()
+
+    if row is not None:
+        model.id = row.id
+        model._Model__exists = True # pylint: disable = protected-access
+
+    model.save ()

update_db.py

@@ -13,24 +13,26 @@ import random
 import string
 import time
 import unicodedata
-from datetime import date, datetime, timedelta
+from datetime import datetime, timedelta
 from typing import Any, TypedDict, cast
 
 import requests
 from eloquent import DatabaseManager, Model
 
-config: dict[str, DbConfig] = { 'mysql': { 'driver':    'mysql',
-                                           'host':      'localhost',
-                                           'database':  'nizika_nico',
-                                           'user':      os.environ['MYSQL_USER'],
-                                           'password':  os.environ['MYSQL_PASS'],
-                                           'prefix':    '' } }
-db = DatabaseManager (config)
-Model.set_connection_resolver (db)
+from models import Comment, Tag, User, Video, VideoHistory, VideoTag
 
 
 def main (
 ) -> None:
+    config: dict[str, DbConfig] = { 'mysql': { 'driver':    'mysql',
+                                               'host':      'localhost',
+                                               'database':  'nizika_nico',
+                                               'user':      os.environ['MYSQL_USER'],
+                                               'password':  os.environ['MYSQL_PASS'],
+                                               'prefix':    '' } }
+    db = DatabaseManager (config)
+    Model.set_connection_resolver (db)
+
     now = datetime.now ()
 
     api_data = search_nico_by_tags (['伊地知ニジカ', 'ぼざろクリーチャーシリーズ'])
@@ -207,167 +209,6 @@ def search_nico_by_tags (
     return result_data
 
 
-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,
-    ) -> 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.__exists = True    # pylint: disable = unused-private-member
-        self.save ()
-
-
-class Tag (Model):
-    id:     int
-    name:   str
-
-    __timestamps__ = False
-
-    @property
-    def video_tags (
-            self,
-    ) -> VideoTag:
-        return self.has_many (VideoTag)
-
-
-class User (Model):
-    id:     int
-    code:   str
-
-    __timestamps__ = False
-
-    @property
-    def comments (
-            self,
-    ) -> 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,
-    ) -> VideoHistory:
-        return self.has_many (VideoHistory)
-
-    @property
-    def video_tags (
-            self,
-    ) -> VideoTag:
-        return self.has_many (VideoTag)
-
-    @property
-    def comments (
-            self,
-    ) -> Comment:
-        return self.has_many (Comment)
-
-    def upsert (
-            self,
-    ) -> None:
-        row = Video.where ('code', self.code).first ()
-        if row is not None:
-            self.id = row.id
-            self.__exists = True    # pylint: disable = unused-private-member
-        self.save ()
-
-
-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,
-    ) -> None:
-        row = (VideoHistory.where ('video_id', self.video_id)
-                           .where ('fetched_at', self.fetched_at)
-                           .first ())
-        if row is not None:
-            self.id = row.id
-            self.__exists = True    # pylint: disable = unused-private-member
-        self.save ()
-
-
-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,
-    ) -> None:
-        row = (VideoTag.where ('video_id', self.video_id)
-                       .where ('tag_id', self.tag_id)
-                       .first ())
-        if row is not None:
-            self.id = row.id
-            self.__exists = True    # pylint: disable = unused-private-member
-        self.save ()
-
-
 class DbConfig (TypedDict):
     driver:     str
     host:       str