paperap.models.abstract package

---

METADATA:

File: __init__.py

Project: paperap

Created: 2025-03-04

Version: 0.0.10

Author: Jess Mann Email: jess@jmann.me

Copyright (c) 2025 Jess Mann

---

LAST MODIFIED:

2025-03-04 By Jess Mann

class paperap.models.abstract.BaseModel(**data)

Bases: BaseModel, ABC

Base model for all Paperless-ngx API objects.

Provides automatic serialization, deserialization, and API interactions with minimal configuration needed.

_meta

Metadata for the model, including filtering and resource information.

_save_lock

Lock for saving operations.

_pending_save

Future object for pending save operations.

Raises:

ValueError – If resource is not provided.

Parameters:

data (Any)

class Meta(model)

Bases: Generic

Metadata for the Model.

name

The name of the model.

read_only_fields

Fields that should not be modified.

filtering_disabled

Fields disabled for filtering.

filtering_fields

Fields allowed for filtering.

supported_filtering_params

Params allowed during queryset filtering.

blacklist_filtering_params

Params disallowed during queryset filtering.

filtering_strategies

Strategies for filtering.

resource

The BaseResource instance.

queryset

The type of QuerySet for the model.

Raises:

ValueError – If both ALLOW_ALL and ALLOW_NONE filtering strategies are set.

Parameters:

model (type[_Self])

__init__(model)

Parameters:

model (type[_Self])

blacklist_filtering_params: ClassVar[set[str]] = {}

field_map: dict[str, str] = {}

filter_allowed(filter_param)

Check if a filter is allowed based on the filtering strategies.

Parameters:

filter_param (str) – The filter parameter to check.

Return type:

bool

Returns:

True if the filter is allowed, False otherwise.

filtering_disabled: ClassVar[set[str]] = {}

filtering_fields: ClassVar[set[str]] = {}

filtering_strategies: ClassVar[set[FilteringStrategies]] = {FilteringStrategies.BLACKLIST}

read_only_fields: ClassVar[set[str]] = {}

save_on_write: bool | None = None

save_timeout: int = ModelPrivateAttr(default=60)

supported_filtering_params: ClassVar[set[str]] = {'limit'}

model: type[TypeVar(_Self, bound= BaseModel)]

name: str

__init__(**data)

Initialize the model with resource and data.

Parameters:

• resource – The BaseResource instance.

• **data (Any) – Additional data to initialize the model.

Raises:

ValueError – If resource is not provided.

classmethod __init_subclass__(**kwargs)

Initialize subclass and set up metadata.

Parameters:

**kwargs (Any) – Additional keyword arguments.

Return type:

None

__str__()

Human-readable string representation.

Return type:

str

Returns:

A string representation of the model.

cleanup()

Clean up resources used by the model class.

Return type:

None

classmethod create(**kwargs)

Create a new model instance.

Parameters:

**kwargs (Any) – Field values to set.

Return type:

Self

Returns:

A new model instance.

Examples

# Create a new Document instance doc = Document.create(filename=”example.pdf”, contents=b”PDF data”)

delete()

Return type:

None

dirty_fields(comparison='both')

Show which fields have changed since last update from the paperless ngx db.

Parameters:

comparison (Literal['saved', 'db', 'both']) – Specify the data to compare (‘saved’ or ‘db’). Db is the last data retrieved from Paperless NGX Saved is the last data sent to Paperless NGX to be saved

Returns:

(original_value, new_value)} of fields that have changed since last update from the paperless ngx db.

Return type:

A dictionary {field

disable_save_on_write()

Disable automatic saving on attribute write.

Return type:

None

enable_save_on_write()

Enable automatic saving on attribute write.

Return type:

None

classmethod from_dict(data)

Create a model instance from API response data.

Parameters:

data (dict[str, Any]) – Dictionary containing the API response data.

Return type:

Self

Returns:

A model instance initialized with the provided data.

Examples

# Create a Document instance from API data doc = Document.from_dict(api_data)

is_dirty(comparison='both')

Check if any field has changed since last update from the paperless ngx db.

Parameters:

comparison (Literal['saved', 'db', 'both']) – Specify the data to compare (‘saved’ or ‘db’). Db is the last data retrieved from Paperless NGX Saved is the last data sent to Paperless NGX to be saved

Return type:

bool

Returns:

True if any field has changed.

abstractmethod is_new()

Check if this model represents a new (unsaved) object.

Return type:

bool

Returns:

True if the model is new, False otherwise.

Examples

# Check if a Document instance is new is_new = doc.is_new()

matches_dict(data)

Check if the model matches the provided data.

Parameters:

data (dict[str, Any]) – Dictionary containing the data to compare.

Return type:

bool

Returns:

True if the model matches the data, False otherwise.

Examples

# Check if a Document instance matches API data matches = doc.matches_dict(api_data)

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'ignore', 'populate_by_name': True, 'use_enum_values': True, 'validate_assignment': True, 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

We need to both initialize private attributes and call the user-defined model_post_init method.

Parameters:

• self (BaseModel)

• context (Any)

Return type:

None

property resource: BaseResource[Self, BaseQuerySet]

property save_executor: ThreadPoolExecutor

should_save_on_write()

Check if the model should save on attribute write, factoring in the client settings.

Return type:

bool

to_dict(*, include_read_only=True, exclude_none=False, exclude_unset=True)

Convert the model to a dictionary for API requests.

Parameters:

• include_read_only (bool) – Whether to include read-only fields.

• exclude_none (bool) – Whether to exclude fields with None values.

• exclude_unset (bool) – Whether to exclude fields that are not set.

Return type:

dict[str, Any]

Returns:

A dictionary with model data ready for API submission.

Examples

# Convert a Document instance to a dictionary data = doc.to_dict()

update(**kwargs)

Update this model with new values.

Subclasses implement this with auto-saving features. However, base BaseModel instances simply call update_locally.

Parameters:

**kwargs (Any) – New field values.

Return type:

None

Examples

# Update a Document instance doc.update(filename=”new_example.pdf”)

update_locally(*, from_db=None, skip_changed_fields=False, **kwargs)

Update model attributes without triggering automatic save.

Parameters:

• **kwargs (Any) – Field values to update

• from_db (bool | None)

• skip_changed_fields (bool)

Return type:

None

Returns:

Self with updated values

class paperap.models.abstract.StandardModel(**data)

Bases: BaseModel, ABC

Standard model for Paperless-ngx API objects with an ID field.

id

Unique identifier for the model.

Parameters:

data (Any)

class Meta(model)

Bases: Meta

Metadata for the StandardModel.

read_only_fields

Fields that should not be modified.

supported_filtering_params

Params allowed during queryset filtering.

Parameters:

model (type[_Self])

blacklist_filtering_params: ClassVar[set[str]] = {}

field_map: dict[str, str] = {}

filtering_disabled: ClassVar[set[str]] = {}

filtering_fields: ClassVar[set[str]] = {'_resource', 'id'}

read_only_fields: ClassVar[set[str]] = {'id'}

supported_filtering_params: ClassVar[set[str]] = {'id', 'id__in', 'limit'}

model: type[_Self]

name: str

__setattr__(name, value)

Override attribute setting to automatically trigger async save.

Parameters:

• name (str) – Attribute name

• value (Any) – New attribute value

Return type:

None

__str__()

Human-readable string representation.

Return type:

str

Returns:

A string representation of the model.

is_new()

Check if this model represents a new (unsaved) object.

Return type:

bool

Returns:

True if the model is new, False otherwise.

Examples

# Check if a Document instance is new is_new = doc.is_new()

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'ignore', 'populate_by_name': True, 'use_enum_values': True, 'validate_assignment': True, 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

We need to both initialize private attributes and call the user-defined model_post_init method.

Parameters:

• self (BaseModel)

• context (Any)

Return type:

None

refresh()

Refresh the model with the latest data from the server.

Return type:

bool

Returns:

True if the model data changes, False on failure or if the data does not change.

Raises:

ResourceNotFoundError – If the model is not found on Paperless. (e.g. it was deleted remotely)

property resource: StandardResource[Self, StandardQuerySet]

save(*, force=False)

Parameters:

force (bool)

Return type:

bool

save_async(*, force=False)

Save this model instance asynchronously.

Changes are sent to the server in a background thread, and the model is updated when the server responds.

Return type:

bool

Returns:

True if the save was successfully submitted async, False otherwise.

Parameters:

force (bool)

save_sync(*, force=False)

Save this model instance synchronously.

Changes are sent to the server immediately, and the model is updated when the server responds.

Return type:

bool

Returns:

True if the save was successful, False otherwise.

Raises:

• ResourceNotFoundError – If the resource doesn’t exist on the server

• RequestError – If there’s a communication error with the server

• PermissionError – If the user doesn’t have permission to update the resource

Parameters:

force (bool)

update(**kwargs)

Update this model with new values and save changes.

NOTE: new instances will not be saved automatically. (I’m not sure if that’s the right design decision or not)

Parameters:

**kwargs (Any) – New field values.

Return type:

None

id: int

class paperap.models.abstract.BaseQuerySet(resource, filters=None, _cache=None, _fetch_all=False, _next_url=None, _last_response=None, _iter=None, _urls_fetched=None)

Bases: Iterable, Generic

A lazy-loaded, chainable query interface for Paperless NGX resources.

BaseQuerySet provides pagination, filtering, and caching functionality similar to Django’s QuerySet. It’s designed to be lazy - only fetching data when it’s actually needed.

Parameters:

• resource (BaseResource[_Model, Self]) – The BaseResource instance.

• filters (dict[str, Any] | None) – Initial filter parameters.

• _cache (list[_Model] | None) – Optional internal result cache.

• _fetch_all (bool) – Whether all results have been fetched.

• _next_url (str | None) – URL for the next page of results.

• _last_response (ClientResponse) – Optional last response from the API.

• _iter (Iterator[_Model] | None) – Optional iterator for the results.

Returns:

A new instance of BaseQuerySet.

Examples

# Create a QuerySet for documents >>> docs = client.documents() >>> for doc in docs: … print(doc.id) 1 2 3

Parameters:

_urls_fetched (list[str] | None)

__bool__()

Return True if the QuerySet has any results.

Return type:

bool

Returns:

True if there are any objects matching the filters

__contains__(item)

Return True if the QuerySet contains the given object.

Parameters:

item (Any) – The object to check for

Return type:

bool

Returns:

True if the object is in the QuerySet

__getitem__(key)

Retrieve an item or slice of items from the QuerySet.

Parameters:

key (int | slice) – An integer index or slice

Return type:

_Model | list[_Model]

Returns:

A single object or list of objects

Raises:

IndexError – If the index is out of range

__init__(resource, filters=None, _cache=None, _fetch_all=False, _next_url=None, _last_response=None, _iter=None, _urls_fetched=None)

Parameters:

• resource (BaseResource[_Model, Self])

• filters (dict[str, Any] | None)

• _cache (list[_Model] | None)

• _fetch_all (bool)

• _next_url (str | None)

• _last_response (ClientResponse)

• _iter (Iterator[_Model] | None)

• _urls_fetched (list[str] | None)

__iter__()

Iterate over the objects in the QuerySet.

Return type:

Iterator[_Model]

Returns:

An iterator over the objects

__len__()

Return the number of objects in the QuerySet.

Return type:

int

Returns:

The count of objects

all()

Return a new QuerySet that copies the current one.

Return type:

Self

Returns:

A copy of the current BaseQuerySet

count()

Return the total number of objects in the queryset.

Return type:

int

Returns:

The total count of objects matching the filters

Raises:

NotImplementedError – If the response does not have a count attribute

count_this_page()

Return the number of objects on the current page.

Return type:

int

Returns:

The count of objects on the current page

Raises:

NotImplementedError – If _last_response is not set

exclude(**kwargs)

Return a new QuerySet excluding objects with the given filters.

Parameters:

**kwargs (Any) – Filters to exclude, where keys are field names and values are excluded values

Return type:

Self

Returns:

A new QuerySet excluding objects that match the filters

Examples

# Get documents with any correspondent except ID 1 docs = client.documents.exclude(correspondent=1)

exists()

Return True if the QuerySet contains any results.

Return type:

bool

Returns:

True if there are any objects matching the filters

filter(**kwargs)

Return a new QuerySet with the given filters applied.

Parameters:

**kwargs (Any) – Filters to apply, where keys are field names and values are desired values. Supports Django-style lookups like field__contains, field__in, etc.

Return type:

Self

Returns:

A new QuerySet with the additional filters applied

Examples

# Get documents with specific correspondent docs = client.documents.filter(correspondent=1)

# Get documents with specific correspondent and document type docs = client.documents.filter(correspondent=1, document_type=2)

# Get documents with title containing “invoice” docs = client.documents.filter(title__contains=”invoice”)

# Get documents with IDs in a list docs = client.documents.filter(id__in=[1, 2, 3])

filter_field_by_str(field, value, *, exact=True, case_insensitive=True)

Filter a queryset based on a given field.

This allows subclasses to easily implement custom filter methods.

Parameters:

• field (str) – The field name to filter by.

• value (str) – The value to filter against.

• exact (bool) – Whether to filter by an exact match.

• case_insensitive (bool) – Whether the filter should be case-insensitive.

Return type:

Self

Returns:

A new QuerySet instance with the filter applied.

first()

Return the first object in the QuerySet, or None if empty.

Return type:

_Model | None

Returns:

The first object or None if no objects match

get(pk)

Retrieve a single object from the API.

Raises NotImplementedError. Subclasses may implement this.

Parameters:

pk (Any) – The primary key (e.g. the id) of the object to retrieve

Return type:

_Model

Returns:

A single object matching the query

Raises:

• ObjectNotFoundError – If no object or multiple objects are found

• NotImplementedError – If the method is not implemented by the subclass

Examples

# Get document with ID 123 doc = client.documents.get(123)

last()

Return the last object in the QuerySet, or None if empty.

Note: This requires fetching all results to determine the last one.

Return type:

_Model | None

Returns:

The last object or None if no objects match

none()

Return an empty QuerySet.

Return type:

Self

Returns:

An empty QuerySet

order_by(*fields)

Return a new QuerySet ordered by the specified fields.

Parameters:

*fields (str) – Field names to order by. Prefix with ‘-’ for descending order.

Return type:

Self

Returns:

A new QuerySet with the ordering applied

Examples

# Order documents by title ascending docs = client.documents.order_by(‘title’)

# Order documents by added date descending docs = client.documents.order_by(‘-added’)

resource: BaseResource[TypeVar(_Model, bound= BaseModel), Self]

filters: dict[str, Any]

class paperap.models.abstract.StandardQuerySet(resource, filters=None, _cache=None, _fetch_all=False, _next_url=None, _last_response=None, _iter=None, _urls_fetched=None)

Bases: BaseQuerySet, Generic

A queryset for StandardModel instances (i.e. BaseModels with standard fields, like id).

Returns:

A new instance of StandardModel.

Raises:

ValueError – If resource is not provided.

Examples

# Create a StandardModel instance model = StandardModel(id=1)

Parameters:

• resource (BaseResource[_Model, Self]) – The BaseResource instance.

• filters (dict[str, Any] | None) – Initial filter parameters.

Returns:

A new instance of StandardQuerySet.

Raises:

ObjectNotFoundError – If no object or multiple objects are found.

Examples

# Create a StandardQuerySet for documents docs = StandardQuerySet(resource=client.documents)

Parameters:

• _cache (list[_Model] | None)

• _fetch_all (bool)

• _next_url (str | None)

• _last_response (ClientResponse)

• _iter (Iterator[_Model] | None)

• _urls_fetched (list[str] | None)

__contains__(item)

Return True if the QuerySet contains the given object.

NOTE: This method only ensures a match by ID, not by full object equality. This is intentional, as the object may be outdated or not fully populated.

Parameters:

item (Any) – The object or ID to check for

Return type:

bool

Returns:

True if the object is in the QuerySet

bulk_action(action, **kwargs)

Perform a bulk action on all objects in the queryset.

This method fetches all IDs in the queryset and passes them to the resource’s bulk_action method.

Parameters:

• action (str) – The action to perform

• **kwargs (Any) – Additional parameters for the action

Return type:

TypeAliasType

Returns:

The API response

Raises:

NotImplementedError – If the resource doesn’t support bulk actions

bulk_assign_correspondent(correspondent_id)

Assign a correspondent to all objects in the queryset.

Parameters:

correspondent_id (int) – Correspondent ID to assign

Return type:

TypeAliasType

Returns:

The API response

bulk_assign_document_type(document_type_id)

Assign a document type to all objects in the queryset.

Parameters:

document_type_id (int) – Document type ID to assign

Return type:

TypeAliasType

Returns:

The API response

bulk_assign_owner(owner_id)

Assign an owner to all objects in the queryset.

Parameters:

owner_id (int) – Owner ID to assign

Return type:

TypeAliasType

Returns:

The API response

bulk_assign_storage_path(storage_path_id)

Assign a storage path to all objects in the queryset.

Parameters:

storage_path_id (int) – Storage path ID to assign

Return type:

TypeAliasType

Returns:

The API response

bulk_assign_tags(tag_ids, remove_existing=False)

Assign tags to all objects in the queryset.

Parameters:

• tag_ids (list[int]) – List of tag IDs to assign

• remove_existing (bool) – If True, remove existing tags before assigning new ones

Return type:

TypeAliasType

Returns:

The API response

bulk_update(**kwargs)

Update all objects in the queryset with the given values.

Parameters:

**kwargs (Any) – Fields to update

Return type:

TypeAliasType

Returns:

The API response

delete()

Delete all objects in the queryset.

Return type:

TypeAliasType

Returns:

The API response

get(pk)

Retrieve a single object from the API.

Parameters:

pk (int) – The ID of the object to retrieve

Return type:

_Model

Returns:

A single object matching the query

Raises:

ObjectNotFoundError – If no object or multiple objects are found

Examples

# Get document with ID 123 doc = client.documents.get(123)

id(value)

Filter models by ID.

Parameters:

value (int | list[int]) – The ID or list of IDs to filter by

Return type:

Self

Returns:

Filtered QuerySet

resource: StandardResource[TypeVar(_Model, bound= StandardModel), Self]

filters: dict[str, Any]

class paperap.models.abstract.FilteringStrategies(*values)

Bases: StrEnum

WHITELIST = 'whitelist'

BLACKLIST = 'blacklist'

ALLOW_ALL = 'allow_all'

ALLOW_NONE = 'allow_none'

class paperap.models.abstract.StatusContext(model, new_status)

Bases: object

Context manager for safely updating model status.

model

The model whose status is being updated.

Type:

SomeModel

new_status

The status to set within the context.

Type:

ModelStatus

previous_status

The status before entering the context.

Type:

ModelStatus

Examples

>>> class SomeModel(BaseModel):
...     def perform_update(self):
...         with StatusContext(self, ModelStatus.UPDATING):
...             # Perform an update

Parameters:

• model (BaseModel)

• new_status (ModelStatus)

__init__(model, new_status)

Parameters:

• model (BaseModel)

• new_status (ModelStatus)

property model: BaseModel

Read-only access to the model.

property new_status: ModelStatus

Read-only access to the new status.

property previous_status: ModelStatus | None

Read-only access to the previous status.

save_lock()

Acquire the save lock

Return type:

None

save_unlock()

Release the save lock, only if this statuscontext previous acquired it.

Return type:

None

Submodules

• paperap.models.abstract.meta module
  • StatusContext
    • StatusContext.model
    • StatusContext.new_status
    • StatusContext.previous_status
    • StatusContext.model
    • StatusContext.new_status
    • StatusContext.previous_status
    • StatusContext.__init__()
    • StatusContext.save_lock()
    • StatusContext.save_unlock()
• paperap.models.abstract.model module
  • ModelConfigType
    • ModelConfigType.populate_by_name
    • ModelConfigType.validate_assignment
    • ModelConfigType.validate_default
    • ModelConfigType.use_enum_values
    • ModelConfigType.extra
    • ModelConfigType.arbitrary_types_allowed
  • BaseModel
    • BaseModel._meta
    • BaseModel._save_lock
    • BaseModel._pending_save
    • BaseModel.Meta
      • BaseModel.Meta.name
      • BaseModel.Meta.read_only_fields
      • BaseModel.Meta.filtering_disabled
      • BaseModel.Meta.filtering_fields
      • BaseModel.Meta.supported_filtering_params
      • BaseModel.Meta.blacklist_filtering_params
      • BaseModel.Meta.filtering_strategies
      • BaseModel.Meta.resource
      • BaseModel.Meta.queryset
      • BaseModel.Meta.name
      • BaseModel.Meta.read_only_fields
      • BaseModel.Meta.filtering_disabled
      • BaseModel.Meta.filtering_fields
      • BaseModel.Meta.supported_filtering_params
      • BaseModel.Meta.blacklist_filtering_params
      • BaseModel.Meta.filtering_strategies
      • BaseModel.Meta.field_map
      • BaseModel.Meta.save_on_write
      • BaseModel.Meta.save_timeout
      • BaseModel.Meta.__init__()
      • BaseModel.Meta.model
      • BaseModel.Meta.filter_allowed()
    • BaseModel.__init_subclass__()
    • BaseModel.model_config
    • BaseModel.__init__()
    • BaseModel.resource
    • BaseModel.save_executor
    • BaseModel.cleanup()
    • BaseModel.model_post_init()
    • BaseModel.from_dict()
    • BaseModel.to_dict()
    • BaseModel.dirty_fields()
    • BaseModel.is_dirty()
    • BaseModel.create()
    • BaseModel.delete()
    • BaseModel.update_locally()
    • BaseModel.update()
    • BaseModel.is_new()
    • BaseModel.should_save_on_write()
    • BaseModel.enable_save_on_write()
    • BaseModel.disable_save_on_write()
    • BaseModel.matches_dict()
    • BaseModel.__str__()
  • StandardModel
    • StandardModel.id
    • StandardModel.id
    • StandardModel.Meta
      • StandardModel.Meta.read_only_fields
      • StandardModel.Meta.supported_filtering_params
      • StandardModel.Meta.read_only_fields
      • StandardModel.Meta.supported_filtering_params
      • StandardModel.Meta.blacklist_filtering_params
      • StandardModel.Meta.field_map
      • StandardModel.Meta.filtering_disabled
      • StandardModel.Meta.filtering_fields
      • StandardModel.Meta.model
      • StandardModel.Meta.name
    • StandardModel.resource
    • StandardModel.update()
    • StandardModel.refresh()
    • StandardModel.save()
    • StandardModel.save_sync()
    • StandardModel.save_async()
    • StandardModel.is_new()
    • StandardModel.__setattr__()
    • StandardModel.__str__()
    • StandardModel.model_config
    • StandardModel.model_post_init()
• paperap.models.abstract.queryset module
  • BaseQuerySet
    • BaseQuerySet.__init__()
    • BaseQuerySet.resource
    • BaseQuerySet.filters
    • BaseQuerySet.filter()
    • BaseQuerySet.exclude()
    • BaseQuerySet.get()
    • BaseQuerySet.count()
    • BaseQuerySet.count_this_page()
    • BaseQuerySet.all()
    • BaseQuerySet.order_by()
    • BaseQuerySet.first()
    • BaseQuerySet.last()
    • BaseQuerySet.exists()
    • BaseQuerySet.none()
    • BaseQuerySet.filter_field_by_str()
    • BaseQuerySet.__iter__()
    • BaseQuerySet.__len__()
    • BaseQuerySet.__bool__()
    • BaseQuerySet.__getitem__()
    • BaseQuerySet.__contains__()
  • StandardQuerySet
    • StandardQuerySet.resource
    • StandardQuerySet.get()
    • StandardQuerySet.id()
    • StandardQuerySet.__contains__()
    • StandardQuerySet.bulk_action()
    • StandardQuerySet.filters
    • StandardQuerySet.delete()
    • StandardQuerySet.bulk_update()
    • StandardQuerySet.bulk_assign_tags()
    • StandardQuerySet.bulk_assign_correspondent()
    • StandardQuerySet.bulk_assign_document_type()
    • StandardQuerySet.bulk_assign_storage_path()
    • StandardQuerySet.bulk_assign_owner()