Version 1.1.5

acacore/__version__.py

@@ -1 +1 @@
-__version__ = "1.1.4"
+__version__ = "1.1.5"

acacore/database/base.py

@@ -541,6 +541,7 @@ def __init__(
         group_by: Optional[list[Union[Column, SelectColumn]]] = None,
         order_by: Optional[list[tuple[Union[str, Column], str]]] = None,
         limit: Optional[int] = None,
+        joins: Optional[list[str]] = None,
     ) -> None:
         """
         A subclass of Table to handle views.
@@ -555,6 +556,7 @@ def __init__(
             order_by: A list tuples containing one column (either as Column or string)
                 and a sorting direction ("ASC", or "DESC").
             limit: The number of rows to limit the results to.
+            joins: Join operations to apply to the view.
         """
         assert columns, "Views must have columns"
         super().__init__(connection, name, columns)
@@ -563,6 +565,7 @@ def __init__(
         self.group_by: list[Union[Column, SelectColumn]] = group_by or []
         self.order_by: Optional[list[tuple[Union[str, Column], str]]] = order_by or []
         self.limit: Optional[int] = limit
+        self.joins: list[str] = joins or []
 
     def __repr__(self) -> str:
         return f'{self.__class__.__name__}("{self.name}", on={self.on!r})'
@@ -577,6 +580,7 @@ def create_statement(self, exist_ok: bool = True) -> str:
         Returns:
             A CREATE VIEW expression.
         """
+        on_table: str = self.on.name if isinstance(self.on, Table) else self.on
         elements: list[str] = ["CREATE VIEW"]
 
         if exist_ok:
@@ -587,13 +591,17 @@ def create_statement(self, exist_ok: bool = True) -> str:
         elements.append("AS")
 
         select_names = [
-            f"{c.name} as {c.alias}" if c.alias else c.name for c in [SelectColumn.from_column(c) for c in self.columns]
+            f"{c.name} as {c.alias}" if c.alias else f"{on_table}.{c.name}"
+            for c in [SelectColumn.from_column(c) for c in self.columns]
         ]
 
         elements.append(
-            f"SELECT {','.join(select_names)} " f"FROM {self.on.name if isinstance(self.on, Table) else self.on}",
+            f"SELECT {','.join(select_names)} " f"FROM {on_table}",
         )
 
+        if self.joins:
+            elements.extend(self.joins)
+
         if self.where:
             elements.append(f"WHERE {self.where}")
 
@@ -682,6 +690,7 @@ def __init__(
         group_by: Optional[list[Union[Column, SelectColumn]]] = None,
         order_by: Optional[list[tuple[Union[str, Column], str]]] = None,
         limit: Optional[int] = None,
+        joins: Optional[list[str]] = None,
     ) -> None:
         """
         A subclass of Table to handle views with models.
@@ -697,6 +706,7 @@ def __init__(
             order_by: A list tuples containing one column (either as Column or string)
                 and a sorting direction ("ASC", or "DESC").
             limit: The number of rows to limit the results to.
+            joins: Join operations to apply to the view.
         """
         super().__init__(
             connection,
@@ -707,6 +717,7 @@ def __init__(
             group_by,
             order_by,
             limit,
+            joins,
         )
         self.model: Type[M] = model
 
@@ -867,6 +878,7 @@ def create_view(
         group_by: Optional[list[Union[Column, SelectColumn]]] = None,
         order_by: Optional[list[tuple[Union[str, Column], str]]] = None,
         limit: Optional[int] = None,
+        joins: Optional[list[str]] = None,
         *,
         select_columns: Optional[list[Union[Column, SelectColumn]]] = None,
     ) -> ModelView[M]:
@@ -882,6 +894,7 @@ def create_view(
         group_by: Optional[list[Union[Column, SelectColumn]]] = None,
         order_by: Optional[list[tuple[Union[str, Column], str]]] = None,
         limit: Optional[int] = None,
+        joins: Optional[list[str]] = None,
     ) -> View:
         ...
 
@@ -894,6 +907,7 @@ def create_view(
         group_by: Optional[list[Union[Column, SelectColumn]]] = None,
         order_by: Optional[list[tuple[Union[str, Column], str]]] = None,
         limit: Optional[int] = None,
+        joins: Optional[list[str]] = None,
         *,
         select_columns: Optional[list[Union[Column, SelectColumn]]] = None,
     ) -> Union[View, ModelView[M]]:
@@ -911,18 +925,9 @@ def create_view(
                 and a sorting direction ("ASC", or "DESC").
             limit: The number of rows to limit the results to.
             select_columns: Optionally, the columns of the view if a model is given and is too limited.
+            joins: Join operations to apply to the view.
         """
         if issubclass(columns, BaseModel):
-            return ModelView[M](
-                self,
-                name,
-                on,
-                columns,
-                select_columns,
-                where,
-                group_by,
-                order_by,
-                limit,
-            )
+            return ModelView[M](self, name, on, columns, select_columns, where, group_by, order_by, limit, joins)
         else:
-            return View(self, name, on, columns, where, group_by, order_by, limit)
+            return View(self, name, on, columns, where, group_by, order_by, limit, joins)

acacore/database/files_db.py

@@ -1,23 +1,49 @@
 from datetime import datetime
 from os import PathLike
+from pathlib import Path
 from sqlite3 import Connection
 from typing import Optional
 from typing import Type
 from typing import Union
 from uuid import UUID
 
+from acacore.models.base import ACABase
 from acacore.models.file import File
 from acacore.models.history import HistoryEntry
-from acacore.models.identification import ChecksumCount
-from acacore.models.identification import SignatureCount
 from acacore.models.metadata import Metadata
+from acacore.models.reference_files import TActionType
 from acacore.utils.functions import or_none
 
+from . import model_to_columns
 from .base import Column
 from .base import FileDBBase
 from .base import SelectColumn
 
 
+class HistoryEntryPath(HistoryEntry):
+    relative_path: Optional[Path] = None
+
+
+class SignatureCount(ACABase):
+    """Signature count datamodel."""
+
+    puid: Optional[str]
+    signature: Optional[str]
+    count: Optional[int]
+
+
+class ChecksumCount(ACABase):
+    """Signature count datamodel."""
+
+    checksum: str
+    count: int
+
+
+class ActionCount(ACABase):
+    action: TActionType
+    count: int
+
+
 class FileDB(FileDBBase):
     def __init__(
         self,
@@ -65,6 +91,16 @@ def __init__(
         self.history = self.create_table("History", HistoryEntry)
         self.metadata = self.create_keys_table("Metadata", Metadata)
 
+        self.history_paths = self.create_view(
+            "_HistoryPaths",
+            self.history,
+            HistoryEntryPath,
+            select_columns=[
+                SelectColumn("Files.relative_path", str, "relative_path"),
+                *model_to_columns(HistoryEntry),
+            ],
+            joins=["left join Files on History.UUID = Files.uuid"],
+        )
         self.identification_warnings = self.create_view(
             "_IdentificationWarnings",
             self.files,
@@ -140,15 +176,45 @@ def __init__(
                 ),
             ],
         )
+        self.actions_count = self.create_view(
+            "_ActionsCount",
+            self.files,
+            ActionCount,
+            None,
+            [
+                Column("action", "varchar", str, str, False, False, False),
+            ],
+            [
+                (Column("count", "int", str, str), "DESC"),
+            ],
+            select_columns=[
+                Column(
+                    "action",
+                    "varchar",
+                    or_none(str),
+                    or_none(str),
+                    False,
+                    False,
+                    False,
+                ),
+                SelectColumn(
+                    f'count("{self.files.name}.action")',
+                    int,
+                    "count",
+                ),
+            ],
+        )
 
     def init(self):
         """Create the default tables and views."""
         self.files.create(True)
         self.history.create(True)
         self.metadata.create(True)
+        self.history_paths.create(True)
         self.identification_warnings.create(True)
         self.checksum_count.create(True)
         self.signature_count.create(True)
+        self.actions_count.create(True)
         self.metadata.update(self.metadata.model())
         self.commit()
 

acacore/models/identification.py

@@ -27,18 +27,3 @@ def check_puid_sig(cls, data: dict[Any, Any]) -> dict[Any, Any]:
             raise ValueError(f"PUID missing for signature {signature}.")
 
         return data
-
-
-class SignatureCount(ACABase):
-    """Signature count datamodel."""
-
-    puid: Optional[str]
-    signature: Optional[str]
-    count: Optional[int]
-
-
-class ChecksumCount(ACABase):
-    """Signature count datamodel."""
-
-    checksum: str
-    count: int

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "acacore"
-version = "1.1.4"
+version = "1.1.5"
 description = ""
 authors = ["Matteo Campinoti <[email protected]>"]
 license = "GPL-3.0"

tests/test_database.py

@@ -12,10 +12,12 @@
 
 # noinspection PyProtectedMember
 from acacore.database.column import _value_to_sql
+from acacore.database.files_db import ActionCount
+from acacore.database.files_db import ChecksumCount
+from acacore.database.files_db import HistoryEntryPath
+from acacore.database.files_db import SignatureCount
 from acacore.models.file import File
 from acacore.models.history import HistoryEntry
-from acacore.models.identification import ChecksumCount
-from acacore.models.identification import SignatureCount
 from acacore.models.reference_files import Action
 
 
@@ -69,12 +71,16 @@ def test_database_classes(database_path: Path):
     assert issubclass(db.history.model, HistoryEntry)
 
     # Check views classes
+    assert isinstance(db.history_paths, ModelView)
+    assert issubclass(db.history_paths.model, HistoryEntryPath)
     assert isinstance(db.identification_warnings, ModelView)
     assert issubclass(db.identification_warnings.model, File)
     assert isinstance(db.checksum_count, ModelView)
     assert issubclass(db.checksum_count.model, ChecksumCount)
     assert isinstance(db.signature_count, ModelView)
     assert issubclass(db.signature_count.model, SignatureCount)
+    assert isinstance(db.actions_count, ModelView)
+    assert issubclass(db.actions_count.model, ActionCount)
 
 
 # noinspection SqlResolve,SqlNoDataSourceInspection
@@ -99,9 +105,11 @@ def test_database_tables(database_path: Path):
     views: list[str] = [
         t for [t] in db.execute("select name from sqlite_master where type = 'view' and name != 'sqlite_master'")
     ]
+    assert db.history_paths.name in views
     assert db.identification_warnings.name in views
     assert db.checksum_count.name in views
     assert db.signature_count.name in views
+    assert db.actions_count.name in views
 
 
 def test_database_columns(database_path: Path):