def back_and_forth(sync_func: Callable,
dbs: tuple[DbConn, DbConn],
- shared: YoutubeVideo | tuple[Any, str]
+ shared: Any
) -> None:
"""Run sync_func on arg pairs + shared, and again with pairs switched."""
host_names = 'local', 'remote'
def sync_objects(host_names: tuple[str, str],
dbs: tuple[DbConn, DbConn],
- shared: tuple[Any, str]
+ cls: Any
) -> None:
"""Equalize both DB's object collections; prefer newer states to older."""
- cls, _ = shared
id_name = 'id_' if 'id' == cls.id_name else cls.id_name
obj_colls = cls.get_all(dbs[0]), cls.get_all(dbs[1])
for obj_0 in [obj for obj in obj_colls[0] if obj not in obj_colls[1]]:
scp.get(PATH_DB, PATH_DB_REMOTE)
with DbConn(PATH_DB) as db_local, DbConn(PATH_DB_REMOTE) as db_remote:
for cls in (QuotaLog, YoutubeQuery, YoutubeVideo, VideoFile):
- back_and_forth(sync_objects, (db_local, db_remote),
- (cls, 'rel_path' if cls is VideoFile else 'id_'))
+ back_and_forth(sync_objects, (db_local, db_remote), cls)
for yt_video_local in YoutubeVideo.get_all(db_local):
back_and_forth(sync_relations, (db_local, db_remote),
yt_video_local)