add store filters

genotype_api/database/base_handler.py

@@ -1,6 +1,10 @@
 from dataclasses import dataclass
+from typing import Type
 
-from sqlalchemy.orm import Session
+import ModelBase
+from sqlalchemy.orm import Session, Query
+
+from genotype_api.database.models import Analysis, Sample
 
 
 @dataclass
@@ -9,3 +13,10 @@ class BaseHandler:
 
     def __init__(self, session: Session):
         self.session = session
+
+    def _get_query(self, table: Type[ModelBase]) -> Query:
+        """Return a query for the given table."""
+        return self.session.query(table)
+
+    def _get_join_analysis_on_sample(self) -> Query:
+        return self._get_query(table=Sample).join(Analysis)

genotype_api/database/crud/read.py

@@ -136,9 +136,6 @@ def get_user_by_id(self, user_id: int) -> User:
     def get_user_by_email(self, email: str) -> User | None:
         return self.session.query(User).filter(User.email == email).first()
 
-    def get_users(self, skip: int = 0, limit: int = 100) -> list[User]:
-        return self.session.query(User).offset(skip).limit(limit).all()
-
     def get_users_with_skip_and_limit(self, skip: int, limit: int) -> list[User]:
         return self.session.query(User).offset(skip).limit(limit).all()
 

genotype_api/database/filters/analysis_filter.py

@@ -0,0 +1,84 @@
+"""Module for the analysis filters."""
+
+from datetime import timedelta, date
+from enum import Enum
+
+from sqlalchemy.orm import Query
+
+from genotype_api.database.models import Analysis
+
+
+def filter_analyses_by_id(analysis_id: int, analyses: Query, **kwargs) -> Query:
+    """Return analysis by internal id."""
+    return analyses.filter(Analysis.id == analysis_id)
+
+
+def filter_analyses_by_type(analysis_type: str, analyses: Query, **kwargs) -> Query:
+    """Return analysis by type."""
+    return analyses.filter(Analysis.analysis_type == analysis_type)
+
+
+def filter_analyses_by_plate_id(plate_id: str, analyses: Query, **kwargs) -> Query:
+    """Return analysis by plate id."""
+    return analyses.filter(Analysis.plate_id == plate_id)
+
+
+def filter_analyses_by_sample_id(sample_id: str, analyses: Query, **kwargs) -> Query:
+    """Return analysis by sample id."""
+    return analyses.filter(Analysis.sample_id == sample_id)
+
+
+def add_skip_and_limit(analyses: Query, skip: int, limit: int) -> Query:
+    """Add skip and limit to the query."""
+    return analyses.offset(skip).limit(limit)
+
+
+def filter_analyses_by_type_between_dates(
+    analyses: Query, analysis_type: str, date_min: date, date_max: date
+) -> Query:
+    """Return analysis by type between dates."""
+    return analyses.filter(
+        Analysis.analysis_type == analysis_type,
+        Analysis.created_at > date_min - timedelta(days=1),
+        Analysis.created_at < date_max + timedelta(days=1),
+    )
+
+
+def apply_analysis_filter(
+    filter_functions: list[callable],
+    analyses: Query,
+    analysis_id: int,
+    analysis_type: str,
+    plate_id: str,
+    sample_id: str,
+    skip: int,
+    limit: int,
+    date_min: date,
+    date_max: date,
+) -> Query:
+    """Apply filtering functions to the analysis queries and return filtered results."""
+
+    for filter_function in filter_functions:
+        analyses: Query = filter_function(
+            analyses=analyses,
+            analysis_id=analysis_id,
+            analysis_type=analysis_type,
+            plate_id=plate_id,
+            sample_id=sample_id,
+            skip=skip,
+            limit=limit,
+            date_min=date_min,
+            date_max=date_max,
+        )
+    return analyses
+
+
+class AnalysisFilter(Enum):
+    """Define Analysis filter functions."""
+
+    BY_ID: callable = filter_analyses_by_id
+    BY_TYPE: callable = filter_analyses_by_type
+    BY_PLATE_ID: callable = filter_analyses_by_plate_id
+    BY_SAMPLE_ID: callable = filter_analyses_by_sample_id
+    SKIP_AND_LIMIT: callable = add_skip_and_limit
+    BY_TYPE_BETWEEN_DATES: callable = filter_analyses_by_type_between_dates

genotype_api/database/filters/plate_filters.py

@@ -0,0 +1,59 @@
+"""Module for the plate filters."""
+
+from sqlalchemy.orm import Query
+
+from genotype_api.database.models import Plate
+
+
+def filter_plates_by_id(entry_id: int, plates: Query, **kwargs) -> Query:
+    """Return plate by internal id."""
+    return plates.filter(Plate.id == entry_id)
+
+
+def filter_plates_by_plate_id(plate_id: str, plates: Query, **kwargs) -> Query:
+    """Return plate by plate id."""
+    return plates.filter(Plate.plate_id == plate_id)
+
+
+def add_skip_and_limit(plates: Query, skip: int, limit: int) -> Query:
+    """Add skip and limit to the query."""
+    return plates.offset(skip).limit(limit)
+
+
+def order_plates(plates: Query, order_by: str, sort_func: callable) -> Query:
+    """Order the plates by the given column."""
+    return plates.order_by(sort_func(order_by))
+
+
+def apply_plate_filter(
+    filter_functions: list[callable],
+    plates: Query,
+    entry_id: int,
+    plate_id: str,
+    skip: int,
+    limit: int,
+    order_by: str,
+    sort_func: callable,
+) -> Query:
+    """Apply filtering functions to the plate queries and return filtered results."""
+
+    for filter_function in filter_functions:
+        plates: Query = filter_function(
+            plates=plates,
+            id=entry_id,
+            plate_id=plate_id,
+            skip=skip,
+            limit=limit,
+            order_by=order_by,
+            sort_func=sort_func,
+        )
+    return plates
+
+
+class PlateFilter:
+    """Define Plate filter functions."""
+
+    BY_ID: callable = filter_plates_by_id
+    BY_PLATE_ID: callable = filter_plates_by_plate_id
+    SKIP_AND_LIMIT: callable = add_skip_and_limit
+    ORDER: callable = order_plates

genotype_api/database/filters/sample_filters.py

@@ -0,0 +1,52 @@
+"""Module for the sample filters."""
+
+from enum import Enum
+from typing import Callable
+
+from sqlalchemy.orm import Query
+
+from genotype_api.database.models import Sample
+
+
+def filter_samples_by_id(sample_id: str, samples: Query, **kwargs) -> Query:
+    """Return sample by internal id."""
+    return samples.filter(Sample.id == sample_id)
+
+
+def filter_samples_contain_id(sample_id: str, samples: Query, **kwargs) -> Query:
+    """Return sample by internal id."""
+    return samples.filter(Sample.id.contains(sample_id))
+
+
+def filter_samples_having_comment(samples: Query, **kwargs) -> Query:
+    """Return sample with a comment."""
+    return samples.filter(Sample.comment.isnot(None))
+
+
+def filter_samples_without_status(samples: Query, **kwargs) -> Query:
+    """Return samples without a status."""
+    return samples.filter(Sample.status.is_(None))
+
+
+def apply_sample_filter(
+    filter_functions: list[Callable],
+    samples: Query,
+    sample_id: str,
+) -> Query:
+    """Apply filtering functions to the sample queries and return filtered results."""
+
+    for filter_function in filter_functions:
+        samples: Query = filter_function(
+            samples=samples,
+            samples_id=sample_id,
+        )
+    return samples
+
+
+class SampleFilter(Enum):
+    """Define Sample filter functions."""
+
+    BY_ID: Callable = filter_samples_by_id
+    CONTAIN_ID: Callable = filter_samples_contain_id
+    HAVING_COMMENT: Callable = filter_samples_having_comment
+    WITHOUT_STATUS: Callable = filter_samples_without_status

genotype_api/database/filters/snp_filters.py

@@ -0,0 +1,41 @@
+"""Module for the SNP filters."""
+
+from sqlalchemy.orm import Query
+
+from genotype_api.database.models import SNP
+
+
+def filter_snps_by_id(snp_id: int, snps: Query, **kwargs) -> Query:
+    """Return SNP by internal id."""
+    return snps.filter(SNP.id == snp_id)
+
+
+def add_skip_and_limit(snps: Query, skip: int, limit: int) -> Query:
+    """Add skip and limit to the query."""
+    return snps.offset(skip).limit(limit)
+
+
+def apply_snp_filter(
+    filter_functions: list[callable],
+    snps: Query,
+    snp_id: int,
+    skip: int,
+    limit: int,
+) -> Query:
+    """Apply filtering functions to the SNP queries and return filtered results."""
+
+    for filter_function in filter_functions:
+        snps: Query = filter_function(
+            snps=snps,
+            snp_id=snp_id,
+            skip=skip,
+            limit=limit,
+        )
+    return snps
+
+
+class SnpFilter:
+    """Define SNP filter functions."""
+
+    BY_ID: callable = filter_snps_by_id
+    SKIP_AND_LIMIT: callable = add_skip_and_limit

genotype_api/database/filters/user_filters.py

@@ -0,0 +1,58 @@
+"""Module for the user filters."""
+
+from enum import Enum
+from typing import Callable
+from sqlalchemy.orm import Query
+from genotype_api.database.models import User
+
+
+def filter_users_by_id(user_id: int, users: Query, **kwargs) -> Query:
+    """Return user by internal id."""
+    return users.filter(User.id == user_id)
+
+
+def filter_users_by_email(email: str, users: Query, **kwargs) -> Query:
+    """Return user by email."""
+    return users.filter(User.email == email)
+
+
+def filter_users_by_name(name: str, users: Query, **kwargs) -> Query:
+    """Return user by name."""
+    return users.filter(User.name == name)
+
+
+def add_skip_and_limit(users: Query, skip: int, limit: int) -> Query:
+    """Add skip and limit to the query."""
+    return users.offset(skip).limit(limit)
+
+
+def apply_user_filter(
+    filter_functions: list[Callable],
+    users: Query,
+    user_id: int,
+    email: str,
+    name: str,
+    skip: int,
+    limit: int,
+) -> Query:
+    """Apply filtering functions to the user queries and return filtered results."""
+
+    for filter_function in filter_functions:
+        users: Query = filter_function(
+            users=users,
+            user_id=user_id,
+            email=email,
+            name=name,
+            skip=skip,
+            limit=limit,
+        )
+    return users
+
+
+class UserFilter(Enum):
+    """Define User filter functions."""
+
+    BY_ID: Callable = filter_users_by_id
+    BY_EMAIL: Callable = filter_users_by_email
+    BY_NAME: Callable = filter_users_by_name
+    SKIP_AND_LIMIT: Callable = add_skip_and_limit