Eloquent 導入（ソースはエラー起きるため main に取込まなぃこと）

update_db.py

@@ -13,17 +13,18 @@ from dataclasses import dataclass
 from datetime import date, datetime, timedelta
 from typing import Any, Type, TypedDict, cast
 
-import mysql.connector
 import requests
-from mysql.connector.connection import MySQLConnectionAbstract
+from eloquent import DatabaseManager, Model
+from eloquent.orm.relations.dynamic_property import DynamicProperty
 
-
+config: dict[str, DbConfig] = { 'mysql': { 'driver':    'mysql',
-class DbNull:
+                                           'host':      'localhost',
-    def __new__ (
+                                           'database':  'nizika_nico',
-            cls,
+                                           'user':      os.environ['MYSQL_USER'],
-    ):
+                                           'password':  os.environ['MYSQL_PASS'],
-        delattr (cls, '__init__')
+                                           'prefix':    '' } }
-DbNullType = Type[DbNull]
+db = DatabaseManager (config)
+Model.set_connection_resolver (db)
 
 
 class VideoSearchParam (TypedDict):
@@ -105,51 +106,165 @@ class VideoTagRow (TypedDict):
     untagged_at:    date | None
 
 
+class DbConfig (TypedDict):
+    driver:     str
+    host:       str
+    database:   str
+    user:       str
+    password:   str
+    prefix:     str
+
+
 def main (
 ) -> None:
-    conn = mysql.connector.connect (user        = os.environ['MYSQL_USER'],
-                                    password    = os.environ['MYSQL_PASS'],
-                                    database    = 'nizika_nico')
-    if not isinstance (conn, MySQLConnectionAbstract):
-        raise TypeError
-
-    now = datetime.now ()
-
-    video_dao = VideoDao (conn)
-    tag_dao = TagDao (conn)
-    video_tag_dao = VideoTagDao (conn)
-    video_history_dao = VideoHistoryDao (conn)
-    comment_dao = CommentDao (conn)
-    user_dao = UserDao (conn)
-
     api_data = search_nico_by_tags (['伊地知ニジカ', 'ぼざろクリーチャーシリーズ'])
 
-    update_tables (video_dao, tag_dao, video_tag_dao, video_history_dao, comment_dao, user_dao,
+    update_tables (api_data, now)
-                   api_data, now)
 
     conn.commit ()
     conn.close ()
 
 
+class Comment (Model):
+    __timestamps__ = False
+
+    @property
+    def video (
+            self,
+    ) -> DynamicProperty:
+        return self.belongs_to (Video)
+
+    @property
+    def user (
+            self,
+    ) -> DynamicProperty:
+        return self.belongs_to (User)
+
+
+class Tag (Model):
+    __timestamps__ = False
+
+    @property
+    def video_tags (
+            self,
+    ) -> DynamicProperty:
+        return self.has_many (VideoTag)
+
+    @property
+    def videos (
+            self,
+    ) -> DynamicProperty:
+        return self.belongs_to_many (Video)
+
+
+class User (Model):
+    __timestamps__ = False
+
+    @property
+    del comments (
+            self,
+    ) -> DynamicProperty:
+        return self.has_many (Comment)
+
+
+class Video (Model):
+    __timestamps__ = False
+
+    @property
+    def video_histories (
+            self,
+    ) -> DynamicProperty:
+        return self.has_many (VideoHistory)
+
+    @property
+    def video_tags (
+            self,
+    ) -> DynamicProperty:
+        return self.has_many (VideoTag)
+
+    @property
+    def tags (
+            self,
+    ) -> DynamicProperty:
+        return self.belongs_to_many (Tag)
+
+    @property
+    def comments (
+            self,
+    ) -> DynamicProperty:
+        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.save ()
+
+
+class VideoHistory (Model):
+    __timestamps__ = False
+
+    @property
+    def video (
+            self,
+    ) -> DynamicProperty:
+        return self.belongs_to (Video)
+
+    def upsert (
+            self,
+    ) -> None:
+        row = (Video
+                .where ('video_id', self.video_id)
+                .where ('fetched_at', self.fetched_at)
+                .first ())
+        if row is not None:
+            self.id = row.id
+        self.save ()
+
+
+class VideoTag (Model):
+    __timestamps__ = False
+
+    @property
+    def video (
+            self,
+    ) -> DynamicProperty:
+        return self.belongs_to (Video)
+
+    @property
+    def tag (
+            self,
+    ) -> DynamicProperty:
+        return self.belongs_to (Tag)
+
+    def upsert (
+            self,
+    ) -> None:
+        row = (Video
+                .where ('video_id', self.video_id)
+                .where ('tag_id', self.tag_id)
+                .first ())
+        if row is not None:
+            self.id = row.id
+        self.save ()
+
+
 def update_tables (
-        video_dao:          VideoDao,
+        api_data:   list[VideoResult],
-        tag_dao:            TagDao,
+        now:        datetime,
-        video_tag_dao:      VideoTagDao,
-        video_history_dao:  VideoHistoryDao,
-        comment_dao:        CommentDao,
-        user_dao:           UserDao,
-        api_data:           list[VideoResult],
-        now:                datetime,
 ) -> None:
     video_ids: list[int] = []
     for datum in api_data:
         tag_names: list[str] = datum['tags'].split ()
-        video = VideoDto (code          = datum['contentId'],
+        video = Video ()
-                          title         = datum['title'],
+        video.code = datum['contentId'],
-                          description   = datum['description'] or '',
+        video.title = datum['title'],
-                          uploaded_at   = datetime.fromisoformat (datum['startTime']))
+        video.description = datum['description'] or '',
-        video_dao.upsert (video, False)
+        video.uploaded_at = datetime.fromisoformat (datum['startTime'])
-        if video.id_ is not None:
+        video.upsert ()
+        if video.id is not None:
             video_ids.append (video.id_)
             video_history = VideoHistoryDto (video_id       = video.id_,
                                              fetched_at     = now,
@@ -298,12 +413,6 @@ def search_nico_by_tags (
 
 
 class VideoDao:
-    def __init__ (
-            self,
-            conn:   MySQLConnectionAbstract,
-    ):
-        self.conn = conn
-
     def find (
             self,
             video_id:               int,
@@ -473,26 +582,7 @@ class VideoDao:
         return video
 
 
-@dataclass (slots = True)
-class VideoDto:
-    code:               str
-    title:              str
-    description:        str
-    uploaded_at:        datetime
-    id_:                int | None                      = None
-    deleted_at:         datetime | DbNullType           = DbNull
-    video_tags:         list[VideoTagDto] | None        = None
-    comments:           list[CommentDto] | None         = None
-    video_histories:    list[VideoHistoryDto] | None    = None
-
-
 class VideoTagDao:
-    def __init__ (
-            self,
-            conn:   MySQLConnectionAbstract,
-    ):
-        self.conn = conn
-
     def fetch_by_video_id (
             self,
             video_id:               int,
@@ -690,24 +780,7 @@ class VideoTagDao:
         return video_tag
 
 
-@dataclass (slots = True)
-class VideoTagDto:
-    video_id:       int
-    tag_id:         int
-    tagged_at:      date
-    id_:            int | None          = None
-    untagged_at:    date | DbNullType   = DbNull
-    video:          VideoDto | None     = None
-    tag:            TagDto | None       = None
-
-
 class TagDao:
-    def __init__ (
-            self,
-            conn:   MySQLConnectionAbstract,
-    ):
-        self.conn = conn
-
     def find (
         self,
         tag_id:                 int,
@@ -783,19 +856,7 @@ class TagDao:
                        name = row['name'])
 
 
-@dataclass (slots = True)
-class TagDto:
-    name:   str
-    id_:    int | None  = None
-
-
 class VideoHistoryDao:
-    def __init__ (
-            self,
-            conn:   MySQLConnectionAbstract,
-    ):
-        self.conn = conn
-
     def fetch_by_video_id (
             self,
             video_id:               int,
@@ -884,22 +945,7 @@ class VideoHistoryDao:
         return video_history
 
 
-@dataclass (slots = True)
-class VideoHistoryDto:
-    video_id:       int
-    fetched_at:     date
-    views_count:    int
-    id_:            int | None      = None
-    video:          VideoDto | None = None
-
-
 class CommentDao:
-    def __init__ (
-            self,
-            conn:   MySQLConnectionAbstract,
-    ):
-        self.conn = conn
-
     def fetch_by_video_id (
             self,
             video_id:               int,
@@ -1001,27 +1047,7 @@ class CommentDao:
         return comment
 
 
-@dataclass (slots = True)
-class CommentDto:
-    video_id:   int
-    comment_no: int
-    user_id:    int
-    content:    str
-    posted_at:  datetime
-    id_:        int | None          = None
-    nico_count: int                 = 0
-    vpos_ms:    int | DbNullType    = DbNull
-    video:      VideoDto | None     = None
-    user:       UserDto | None      = None
-
-
 class UserDao:
-    def __init__ (
-            self,
-            conn:   MySQLConnectionAbstract,
-    ):
-        self.conn = conn
-
     def fetch_by_code (
             self,
             user_code:  str
@@ -1062,11 +1088,5 @@ class UserDao:
                         code    = row['code'])
 
 
-@dataclass (slots = True)
-class UserDto:
-    code:   str
-    id_:    int | None  = None
-
-
 if __name__ == '__main__':
     main ()