Implemented PhotoInfo.owner, AlbumInfo.owner, #216, #239

osxphotos/_version.py

@@ -1,3 +1,4 @@
 """ version info """
 
-__version__ = "0.42.84"
+__version__ = "0.42.85"
+

osxphotos/albuminfo.py

@@ -22,6 +22,7 @@ from ._constants import (
     AlbumSortOrder,
 )
 from .datetime_utils import get_local_tz
+from .query_builder import get_query
 
 
 def sort_list_by_keys(values, sort_keys):
@@ -131,6 +132,22 @@ class AlbumInfoBaseClass:
     def photos(self):
         return []
 
+    @property
+    def owner(self):
+        """Return name of photo owner for shared album (Photos 5+ only), or None if not shared"""
+        if self._db._db_version <= _PHOTOS_4_VERSION:
+            return None
+
+        try:
+            return self._owner
+        except AttributeError:
+            query = get_query(
+                "cloud_album_owner", photos_ver=self._db._photos_ver, uuid=self.uuid
+            )
+            result = self._db.execute(query).fetchone()
+            self._owner = result[0] if result else None
+            return self._owner
+
     def __len__(self):
         """return number of photos contained in album"""
         return len(self.photos)

osxphotos/cli.py

@@ -4103,7 +4103,11 @@ def repl(ctx, cli_obj, db):
     get_photo = photosdb.get_photo
     show = _show_photo
     get_selected = _get_selected(photosdb)
-    selected = get_selected()
+    try:
+        selected = get_selected()
+    except Exception:
+        # get_selected sometimes fails
+        selected = []
 
     def inspect(obj):
         """inspect object"""

osxphotos/photoinfo/photoinfo.py

@@ -38,6 +38,7 @@ from ..albuminfo import AlbumInfo, ImportInfo
 from ..personinfo import FaceInfo, PersonInfo
 from ..phototemplate import PhotoTemplate, RenderOptions
 from ..placeinfo import PlaceInfo4, PlaceInfo5
+from ..query_builder import get_query
 from ..text_detection import detect_text
 from ..uti import get_preferred_uti_extension, get_uti_for_extension
 from ..utils import _debug, _get_resource_loc, findfiles
@@ -1098,6 +1099,22 @@ class PhotoInfo:
             logging.warning(f"Did not find signature for {self.uuid} in _db_signatures")
         return duplicates
 
+    @property
+    def owner(self):
+        """Return name of photo owner for shared photos (Photos 5+ only), or None if not shared"""
+        if self._db._db_version <= _PHOTOS_4_VERSION:
+            return None
+
+        try:
+            return self._owner
+        except AttributeError:
+            query = get_query(
+                "shared_owner", photos_ver=self._db._photos_ver, uuid=self.uuid
+            )
+            result = self._db.execute(query).fetchone()
+            self._owner = result[0] if result else None
+            return self._owner
+
     def render_template(
         self, template_str: str, options: Optional[RenderOptions] = None
     ):

osxphotos/photosdb/photosdb.py

@@ -330,6 +330,8 @@ class PhotosDB:
         else:
             self._process_database5()
 
+        self._db_connection, _ = self.get_db_connection()
+
     @property
     def keywords_as_dict(self):
         """return keywords as dict of keyword, count in reverse sorted order (descending)"""
@@ -790,8 +792,8 @@ class PhotosDB:
                 "creation_date": album[8],
                 "start_date": None,  # Photos 5 only
                 "end_date": None,  # Photos 5 only
-                "customsortascending": None, # Photos 5 only
-                "customsortkey": None, # Photos 5 only
+                "customsortascending": None,  # Photos 5 only
+                "customsortkey": None,  # Photos 5 only
             }
 
         # get details about folders
@@ -1104,7 +1106,9 @@ class PhotosDB:
             # get info on special types
             self._dbphotos[uuid]["specialType"] = row[25]
             self._dbphotos[uuid]["masterModelID"] = row[26]
-            self._dbphotos[uuid]["pk"] = row[26] # same as masterModelID, to match Photos 5
+            self._dbphotos[uuid]["pk"] = row[
+                26
+            ]  # same as masterModelID, to match Photos 5
             self._dbphotos[uuid]["panorama"] = True if row[25] == 1 else False
             self._dbphotos[uuid]["slow_mo"] = True if row[25] == 2 else False
             self._dbphotos[uuid]["time_lapse"] = True if row[25] == 3 else False
@@ -3354,6 +3358,10 @@ class PhotosDB:
 
         return photos
 
+    def execute(self, sql):
+        """Execute sql statement and return cursor"""
+        return self._db_connection.cursor().execute(sql)
+
     def _duplicate_signature(self, uuid):
         """Compute a signature for finding possible duplicates"""
         return (
@@ -3381,6 +3389,10 @@ class PhotosDB:
         """
         return len(self._dbphotos)
 
+    def __del__(self):
+        if getattr(self, "_db_connection", None):
+            self._db_connection.close()
+
 
 def _get_photos_by_attribute(photos, attribute, values, ignore_case):
     """Search for photos based on values being in PhotoInfo.attribute

osxphotos/queries/README.md

@@ -0,0 +1,5 @@
+# Query templates
+
+This folder contains sql query templates for getting various photo properties
+
+The query templates must be rendered with mako (see query_builder.py)

osxphotos/queries/cloud_album_owner.sql.mako

@@ -0,0 +1,4 @@
+-- Get owner name for shared iCloud album
+SELECT ZGENERICALBUM.ZCLOUDOWNERFULLNAME AS OWNER_FULLNAME
+FROM ZGENERICALBUM
+WHERE ZGENERICALBUM.ZUUID = '${uuid}'

osxphotos/queries/shared_owner.sql.mako

@@ -0,0 +1,25 @@
+-- Get the owner name of person who owns a photo in a shared album
+WITH case1 AS
+(
+    -- Case where someone has invited you to a shared album
+    -- Need to get the owner of the shared album
+    SELECT ZGENERICALBUM.ZCLOUDOWNERFULLNAME as OWNER_FULLNAME
+    FROM ZGENERICALBUM
+    JOIN ${asset_table} ON ${asset_table}.ZCLOUDOWNERHASHEDPERSONID = ZGENERICALBUM.ZCLOUDOWNERHASHEDPERSONID
+    WHERE ${asset_table}.ZUUID = "${uuid}"
+),
+case2 AS
+(
+    -- Case where you have invited someone to a shared album
+    -- Need to get the data for person who was invited to the album
+    SELECT
+    ZCLOUDSHAREDALBUMINVITATIONRECORD.ZINVITEEFULLNAME AS OWNER_FULLNAME
+    FROM ZCLOUDSHAREDALBUMINVITATIONRECORD 
+    JOIN ${asset_table} ON ${asset_table}.ZCLOUDOWNERHASHEDPERSONID = ZCLOUDSHAREDALBUMINVITATIONRECORD.ZINVITEEHASHEDPERSONID
+    WHERE ${asset_table}.ZUUID = "${uuid}"
+    ORDER BY ZCLOUDSHAREDALBUMINVITATIONRECORD.Z_PK
+    LIMIT 1
+)
+SELECT * FROM case1
+UNION
+SELECT * FROM case2 WHERE NOT EXISTS (SELECT * FROM case1)

osxphotos/queries/title.sql.mako

@@ -0,0 +1,6 @@
+-- Get title of a photo with given UUID
+SELECT 
+ZADDITIONALASSETATTRIBUTES.ZTITLE
+FROM ZADDITIONALASSETATTRIBUTES
+JOIN ${asset_table} ON ${asset_table}.Z_PK = ZADDITIONALASSETATTRIBUTES.ZASSET
+WHERE ${asset_table}.ZUUID = "${uuid}"

osxphotos/query_builder.py

@@ -0,0 +1,36 @@
+"""Build sql queries from template to retrieve info from the database"""
+
+import os.path
+import pathlib
+from functools import lru_cache
+
+from mako.template import Template
+
+from ._constants import _DB_TABLE_NAMES
+
+QUERY_DIR = os.path.join(os.path.dirname(__file__), "queries")
+
+
+def get_query(query_name, photos_ver, **kwargs):
+    """Return sqlite query string for an attribute and a given database version"""
+
+    # there can be a single query for multiple database versions or separate queries for each version
+    # try generic version first (most common case), if that fails, look for version specific query
+    query_string = _get_query_string(query_name, photos_ver)
+    asset_table = _DB_TABLE_NAMES[photos_ver]["ASSET"]
+    query_template = Template(query_string)
+    return query_template.render(asset_table=asset_table, **kwargs)
+
+
+@lru_cache(maxsize=None)
+def _get_query_string(query_name, photos_ver):
+    """Return sqlite query string for an attribute and a given database version"""
+    query_file = pathlib.Path(QUERY_DIR) / f"{query_name}.sql.mako"
+    if not query_file.is_file():
+        query_file = pathlib.Path(QUERY_DIR) / f"{query_name}_{photos_ver}.sql.mako"
+        if not query_file.is_file():
+            raise FileNotFoundError(f"Query file '{query_file}' not found")
+
+    with open(query_file, "r") as f:
+        query_string = f.read()
+    return query_string