application.db package#

Submodules#

application.db.apikeys module#

This module allows for creating, validating, and managing API keys in the database.

application.db.apikeys.delete_api_key(key)#

Deletes an API key from the database.

Parameters:

key (str) – The API key to be deleted.

Returns:

True if the API key was successfully deleted, False otherwise.

Return type:

bool

application.db.apikeys.get_api_keys()#

Retrieve a list of API keys from the database.

This function queries the database to find API keys, sorts them by the ‘created’ field in descending order, and limits the result to 200 entries.

Returns:

A list of API keys sorted by creation date in descending order.

Return type:

list

application.db.apikeys.new_api_key(description, permissions)#

Generate a new API key with the given description and permissions.

Parameters:

  • description (str) – A description for the new API key.

  • permissions (list[str]) – A list of permissions associated with the API key.

Returns:

The generated API key.

Return type:

str

application.db.apikeys.valid_api_key(key)#

Checks if the provided API key is valid by searching for it in the database.

Parameters:

key (str) – The API key to be validated.

Returns:

True if the API key is found in the database, False otherwise.

Return type:

bool

application.db.blob module#

This module provides functions to manage blobs in the database, including creating, updating, deleting, and retrieving blob data. It also handles file uploads, packing and unpacking of ZIP files, and generating previews for various file types.

application.db.blob.add_reference(id)#

Increment the ‘references’ field of a document in the database by 1.

Parameters:

id (str) – The unique identifier of the document to update.

Return type:

None

Returns:

None

application.db.blob.build_blob_query(filter, user_id)#

Builds a MongoDB query for searching blobs based on the provided filter and user ID.

Parameters:

  • filter (BlobSearchFilter) – A filter object containing various search criteria.

  • user_id (ObjectId) – The ID of the user making the query.

Returns:

A MongoDB query dictionary constructed based on the provided filter and user ID.

Return type:

dict

The filter object can contain the following keys:

  • tag_expr (str): A tag expression to filter blobs by tags.

  • creator (str or list): A username or a list of usernames to filter blobs by their creators.

  • begin_date (datetime): The start date to filter blobs created after this date.

  • end_date (datetime): The end date to filter blobs created before this date.

  • name (str): A regex pattern to filter blobs by their names.

  • ephemeral (bool): A boolean to filter blobs by their ephemeral status.

The query will always include blobs that are either not hidden or created by the user.

application.db.blob.cancel_zip(blob_zip_id)#

Cancels the zip operation for the given blob_zip_id.

Parameters:

blob_zip_id (str) – The ID of the blob zip operation to cancel.

Raises:

BlobDoesNotExistError – If the blob_zip_id does not exist in the _zip_progress.

Returns:

A dictionary containing the progress and item of the zip operation.

Return type:

dict

application.db.blob.count_blobs(filter, user_id)#

Count the number of blobs in the database that match the given filter and user ID.

Parameters:

  • filter (BlobSearchFilter) – The filter criteria to apply to the blob search.

  • user_id (ObjectId) – The ID of the user whose blobs are being counted.

Returns:

The number of blobs that match the filter criteria for the given user.

Return type:

int

application.db.blob.count_tag_uses(tag, user_ids)#

Count the number of documents in the database that contain a specific tag and are created by any of the specified users.

Parameters:

  • tag (str) – The tag to search for in the documents.

  • user_ids (list[str]) – A list of user identifiers to filter the documents by their creators.

Returns:

The count of documents that match the specified tag and creators.

Return type:

int

application.db.blob.create_blob(name, tags, hidden=False, ephemeral=False)#

Creates a new blob entry in the database.

Parameters:

  • name (str) – The name of the blob, typically the filename.

  • tags (list[str]) – A list of tags associated with the blob. Defaults to an empty list.

  • hidden (bool, optional) – A flag indicating if the blob should be hidden. Defaults to False.

  • ephemeral (bool, optional) – A flag indicating if the blob is ephemeral. Defaults to False.

Returns:

A tuple containing the inserted blob’s ID and the file extension.

Return type:

tuple[str, str]

application.db.blob.create_blob_previews(uploaded_blobs)#

Generates and stores preview and thumbnail images for uploaded blobs based on their file type.

For each blob in the provided list: - If the blob is an image, creates a preview (512px) and a thumbnail (128px). - If the blob is a 3D model, generates a preview that can be viewed in browser. - If the blob is a video, generates a preview image. - If the blob is a PDF, creates a preview image of the first page.

Parameters:

uploaded_blobs (list[dict[str,str]]) – A list of dictionaries containing blob IDs and extensions.

Return type:

None

Returns:

None

application.db.blob.create_preview_model(path, preview_id)#

Creates a preview model from a blob and updates the database with the preview information.

Parameters:

  • path (str) – The file path to the blob.

  • preview_id (str) – The unique identifier for the preview.

Return type:

None

Returns:

None

application.db.blob.create_preview_video(path, preview_id)#

Creates a preview video from the first frame of the given video blob and updates the database with the preview information.

Parameters:

  • path (str) – The file path to the video blob.

  • preview_id (str) – The unique identifier for the preview.

Return type:

None

Returns:

None

application.db.blob.delete_blob(blob_id)#

Deletes a blob from the database and the file system.

Parameters:

blob_id (str) – The ID of the blob to be deleted.

Returns:

The data of the deleted blob.

Return type:

dict

Raises:

BlobDoesNotExistError – If the blob with the given ID does not exist.

application.db.blob.file_info(filename)#

Calculate the MD5 checksum and size of a file.

Parameters:

filename (str) – The path to the file.

Returns:

A tuple containing the size of the file in bytes (int) and the MD5 checksum (str).

Return type:

tuple

application.db.blob.get_blob_data(id)#

Retrieve blob data from the database by its ID.

Parameters:

id (str) – The ID of the blob to retrieve.

Returns:

A dictionary containing the blob data. If the blob exists, the dictionary

will include the blob’s ID and the creator’s username. If the creator does not exist, the creator field will contain the creator’s ID as a string.

Return type:

dict

Raises:

exceptions.BlobDoesNotExistError – If the blob with the given ID does not exist.

application.db.blob.get_blobs(filter, start, count, sorting, user_id)#

Retrieve a list of blobs from the database based on the provided filter, pagination, and sorting options.

Parameters:

  • filter (BlobSearchFilter) – The filter criteria to apply to the blob search.

  • start (int) – The starting index for pagination.

  • count (int) – The number of blobs to retrieve.

  • sorting (Sorting) – The sorting options for the blobs.

  • user_id (ObjectId) – The ID of the user performing the search.

Returns:

A list of blobs matching the search criteria.

Return type:

list

application.db.blob.get_tags_from_mime(mime)#

Extracts tags from a MIME type string.

Parameters:

mime (str) – The MIME type string to be split into tags.

Returns:

A list of tags obtained by splitting the MIME type string by ‘/’.

Return type:

list

application.db.blob.get_uid()#

Generate a unique identifier (UID).

This function generates a new unique identifier using the UUID version 4 standard, which creates a random UUID.

Returns:

A string representation of the generated UUID.

Return type:

str

application.db.blob.get_zip_progress(blob_zip_id)#

Retrieve the progress of a zip operation by its ID.

Parameters:

blob_zip_id (str) – The ID of the zip operation.

Returns:

A dictionary containing the progress and the current item being processed.

Return type:

dict

Raises:

BlobDoesNotExistError – If the provided blob_zip_id does not exist in the progress tracking.

application.db.blob.init()#

Initializes the blob module by setting the global blob_path and deleting all ephemeral files that are not referred to by any data.

Note

This function should be called on startup, and regular restarts should be scheduled to ensure that ephemeral files are cleaned up.

Return type:

None

application.db.blob.mark_as_completed(id, size, md5sum)#

Marks a blob as completed in the database.

Parameters:

  • id (str) – The unique identifier of the blob.

  • size (int) – The size of the blob in bytes.

  • md5sum (str) – The MD5 checksum of the blob.

Return type:

None

Returns:

None

application.db.blob.remove_reference(id)#

Decrements the reference count of a database object identified by the given ID.

If the provided ID is not a valid ObjectId, the function returns immediately. Otherwise, it decrements the ‘references’ field of the object by 1 and ensures that the ‘references’ field does not go below 0.

Parameters:

id (str) – The ID of the database object whose reference count is to be decremented.

Return type:

None

Returns:

None

application.db.blob.save_blob_data(file, auto_unzip, tags, hidden=False, ephemeral=False)#

Save blob data to storage and optionally unzip if the file is a zip archive.

Parameters:

  • file (FileStorage) – The file to be saved.

  • auto_unzip (bool) – If True, automatically unzip the file if it is a zip archive.

  • tags (list[str]) – List of tags to associate with the blob.

  • hidden (bool, optional) – If True, mark the blob as hidden. Defaults to False.

  • ephemeral (bool, optional) – If True, mark the blob as ephemeral. Defaults to False.

Returns:

A list of dictionaries containing the unique ID and file extension of the uploaded blobs.

Return type:

list

application.db.blob.set_blob_ephemeral(blob_id, ephemeral)#

Sets the ‘ephemeral’ status of a blob in the database and returns the updated blob data.

Parameters:

  • blob_id (str) – The unique identifier of the blob.

  • ephemeral (bool) – The new ephemeral status to set for the blob.

Returns:

The updated blob data with the new ephemeral status.

Return type:

dict

application.db.blob.set_blob_hidden(blob_id, hidden)#

Set the hidden status of a blob in the database.

Parameters:

  • blob_id (str) – The unique identifier of the blob.

  • hidden (bool) – The hidden status to set for the blob.

Returns:

The updated blob data.

Return type:

dict

Raises:

BlobDoesNotExistError – If the blob with the given ID does not exist.

application.db.blob.set_blob_tags(blob_id, tags)#

Set tags for a blob in the database.

This function updates the tags for a blob identified by its ID. If the blob does not exist, it raises a BlobDoesNotExistError. The tags are converted to lowercase and duplicates are removed.

Parameters:

  • blob_id (str) – The ID of the blob to update.

  • tags (list) – A list of tags to set for the blob.

Returns:

The updated blob data with the new tags.

Return type:

dict

Raises:

BlobDoesNotExistError – If the blob with the specified ID does not exist.

application.db.blob.set_mime_from_ext(mime, ext)#

Sets the MIME type based on the file extension.

Parameters:

  • mime (str) – The initial MIME type.

  • ext (str) – The file extension.

Returns:

The updated MIME type based on the file extension.

Return type:

str

application.db.blob.sum_blob_size(filter, user_id)#

Calculate the total size of blobs that match the given filter for a specific user.

Parameters:

  • filter (BlobSearchFilter) – The filter criteria to search for blobs.

  • user_id (ObjectId) – The ID of the user whose blobs are being queried.

Returns:

The total size of the blobs that match the filter.

Returns 0 if the user does not exist or no blobs match the filter.

Return type:

int

application.db.blob.unzip_file_into_blobs(filename, tags, hidden, ephemeral)#

Unzips a ZIP archive and creates blobs from each file inside.

Parameters:

  • filename (str) – Path to the ZIP file to be extracted.

  • tags (list[str]) – List of tags to associate with each created blob.

  • hidden (bool) – Whether the created blobs should be marked as hidden.

  • ephemeral (bool) – Whether the created blobs should be marked as ephemeral.

Returns:

A list of dictionaries, each containing the ‘id’ and ‘ext’ of a created blob.

Return type:

list

application.db.blob.zip_matching_blobs(filter, user_id, blob_zip_id)#

Create a ZIP archive of blobs that match the given filter and user ID.

Parameters:

  • filter (BlobSearchFilter) – The filter criteria to match blobs.

  • user_id (ObjectId) – The ID of the user requesting the ZIP archive.

  • blob_zip_id (str) – A unique identifier for the ZIP archive.

Returns:

Information about the created ZIP blob, including its ID.

Return type:

dict

Raises:

exceptions.BlobDoesNotExistError – If the created ZIP blob does not exist in the database.

application.db.book module#

application.db.book

application.db.book.append_ebook(book_id, ebook_url)#

Append an ebook to the book’s ebook list and update the database.

Parameters:

  • book_id (str) – The unique identifier of the book.

  • ebook_url (str) – The URL of the ebook to be appended.

Returns:

The updated book data.

Return type:

dict

Raises:

exceptions.BlobDoesNotExistError – If the ebook URL does not exist in the blob storage.

application.db.book.borrow_book(book_id, user_data)#

Allows a user to borrow a book by updating the book’s share history and status.

Parameters:

  • book_id (str) – The ID of the book to be borrowed.

  • user_data (dict) – A dictionary containing user information, including: - ‘_id’: The user’s unique identifier. - ‘username’: The user’s name.

Returns:

The data of the book being borrowed.

Return type:

dict

Raises:
application.db.book.build_book_query(filter, sort)#

Builds a MongoDB query and aggregation pipeline based on the provided filter and sort criteria.

Parameters:

  • filter (BookSearchFilter) – A dictionary containing search filters for books. - owner (str or list): The owner(s) of the book. - author (str): The author of the book. - genre (str): The genre of the book. - shared (bool): Whether the book is shared. - title (str): The title or ISBN of the book.

  • sort (list, optional) – A list of tuples specifying the sort order. Defaults to [].

Returns:

A tuple containing:

  • aggregate (list or None): The aggregation pipeline if applicable, otherwise None.

  • query (dict): The MongoDB query dictionary.

Return type:

tuple

application.db.book.build_keywords(book_data)#

Extracts and builds a list of unique keywords from the given book data.

This function processes specific fields (‘title’, ‘subtitle’, ‘description’, ‘authors’) from the provided book data dictionary. It tokenizes the content of these fields into keywords and returns a sorted list of unique keywords.

Parameters:

book_data (dict) – A dictionary containing book information. Expected keys are ‘title’, ‘subtitle’, ‘description’, and ‘authors’. The value for ‘authors’ can be a list of strings, while the other fields are expected to be strings.

Returns:

A sorted list of unique keywords extracted from the specified fields

in the book data.

Return type:

list[str]

application.db.book.count_all_user_books(filter_users=None)#

Count the number of books owned by each user.

This function aggregates the total number of books for each user in the database. It can also filter the results to include only specific users if a list of user IDs is provided.

Parameters:

filter_users (list | None) – A list of user IDs to filter the results. If None, all users are included.

Returns:

A list of dictionaries, each containing the ‘owner’ information

(username and display name) and the ‘count’ of books owned by that user.

Return type:

list

application.db.book.count_books(filter)#

Count the number of books in the database based on the provided filter.

Parameters:

filter (BookSearchFilter) – The filter criteria to search for books.

Returns:

The count of books that match the filter criteria.

Returns 0 if the user does not exist or no books match the criteria.

Return type:

int

Raises:

exceptions.UserDoesNotExistError – If the user specified in the filter does not exist.

application.db.book.create_book(owner, data)#

Creates a new book entry in the database.

Parameters:

  • owner (str) – The username of the book owner.

  • data (dict) – A dictionary containing book information. Expected keys are: - ‘rfid’: The RFID of the book. - ‘title’: The title of the book. - ‘subtitle’: The subtitle of the book. - ‘authors’: A list of authors of the book. - ‘publisher’: The publisher of the book. - ‘publishedDate’: The published date of the book. - ‘description’: A description of the book. - ‘pageCount’: The number of pages in the book. - ‘thumbnail’: A URL to the thumbnail image of the book. - ‘isbn’ (optional): The ISBN of the book.

Returns:

A dictionary containing the created book data.

Return type:

dict

application.db.book.edit_book(id, new_data)#

Edit the details of a book in the database.

Parameters:

  • id (str) – The unique identifier of the book to be edited.

  • new_data (dict) – A dictionary containing the new data for the book.

Returns:

The updated book data.

Return type:

dict

This function retrieves the current data of the book using its ID, compares it with the new data provided, and updates the fields that have changed. It also updates the ‘noSyncFields’ to include the changed fields. Finally, it updates the book data in the database and returns the updated book data.

application.db.book.get_book(id, *, parse=False)#

Retrieve a book from the database by its ID.

Parameters:

  • id (str) – The ID of the book to retrieve.

  • parse (bool, optional) – If True, the book data will be processed before returning. Defaults to False.

Returns:

The book data, either raw or processed based on the parse parameter.

Return type:

dict

Raises:

BookTagDoesNotExistError – If no book with the given ID is found in the database.

application.db.book.get_book_tag(rfid, *, parse=False)#

Retrieve a book tag from the database using its RFID.

Parameters:

  • rfid (str) – The RFID of the book tag to retrieve.

  • parse (bool, optional) – If True, the book data will be processed before returning. Defaults to False.

Returns:

The book tag data. If parse is True, the data will be processed.

Return type:

dict

Raises:

BookTagDoesNotExistError – If no book tag with the given RFID is found in the database.

application.db.book.get_books(filter, start, count, sorting)#

Retrieve a list of books from the database based on the given filter, pagination, and sorting criteria.

Parameters:

  • filter (BookSearchFilter) – The filter criteria to apply to the book search.

  • start (int) – The starting index for pagination.

  • count (int) – The number of books to retrieve.

  • sorting (Sorting) – The sorting criteria, including fields to sort by and order (ascending/descending).

Returns:

A list of books that match the filter criteria, sorted and paginated as specified.

Return type:

list

application.db.book.init()#

Initialize the application by pre-caching some data.

Return type:

None

application.db.book.keyword_tokenize(text)#

Tokenizes the input text into a list of unique keywords.

Parameters:

text (str) – The input text to be tokenized.

Returns:

A list of unique keywords extracted from the input text.

Return type:

list[str]

Links a book tag to a book in the database.

Parameters:

  • owner (str) – The username of the owner of the book.

  • rfid (str) – The RFID tag to be linked to the book.

  • book_id (str) – The ID of the book to be linked.

Returns:

A dictionary containing the book data after linking the tag.

Return type:

dict

Raises:

BookTagExistsError – If the RFID tag is already linked to a book.

application.db.book.next_out_of_date_book_rfid(before)#

Retrieve the RFID of the next book that is out of date.

This function queries the database for a book whose ‘lastSync’ field is earlier than the specified ‘before’ datetime. If such a book is found, its RFID is returned. If no such book is found, None is returned.

Parameters:

before (datetime) – The cutoff datetime to find books that are out of date.

Returns:

The RFID of the next out-of-date book, or None if no such book is found.

Return type:

str

application.db.book.norm_query(query, ownerq)#

Normalize a query by combining it with an owner query if provided.

Parameters:

  • query (dict) – The initial query dictionary.

  • ownerq (dict | None) – An optional owner query dictionary.

Returns:

The combined query dictionary if both query and ownerq are provided,

otherwise returns the non-None query dictionary.

Return type:

dict

application.db.book.process_book_tag(book_data)#

Processes and enriches book data with additional information.

Parameters:

book_data (dict) – A dictionary containing book information.

Returns:

The enriched book data dictionary.

Return type:

dict

Raises:
application.db.book.process_share_hist(share_history)#

Processes a list of share history records and updates the display name for each record.

Parameters:

share_history (list) – A list of dictionaries, where each dictionary represents a share history record. Each record must contain ‘user_id’ and ‘name’ keys.

Returns:

A list of dictionaries with updated ‘display_name’ keys.

If ‘user_id’ is None, ‘display_name’ is set to ‘name’. If ‘user_id’ is not None, ‘display_name’ is set to the display name of the user. If the user does not exist, ‘display_name’ is set to ‘name’.

Return type:

list

application.db.book.refresh_book_data(rfid)#

Refreshes the book data in the database using the provided RFID.

This function retrieves the book data from the database using the RFID, fetches updated book information from the Google Books API, and updates the database with the new information. If the book data does not exist in the database, it raises a BookTagDoesNotExistError.

Parameters:

rfid (str) – The RFID of the book to refresh.

Raises:

BookTagDoesNotExistError – If the book data does not exist in the database.

Return type:

None

Change the RFID of a book tag to a new one.

Parameters:

  • id (str) – The ID of the book.

  • rfid (str) – The new RFID for the book tag.

Returns:

The updated book tag information.

Return type:

dict

Raises:
application.db.book.remove_ebook(book_id, index)#

Remove an ebook from the book’s ebook list and update the database.

Parameters:

  • book_id (str) – The unique identifier of the book.

  • index (int) – The URL of the ebook to be appended.

Returns:

The updated book data.

Return type:

dict

Raises:

exceptions.BookTagDoesNotExistError – If the book with the given ID does not exist.

application.db.book.return_book(book_id, user_data)#

Return a borrowed book to the library.

Parameters:

  • book_id (str) – The ID of the book to be returned.

  • user_data (dict) – The data of the user returning the book, containing at least the user’s ID.

Returns:

The data of the book being returned.

Return type:

dict

Raises:
application.db.book.set_book_owner(id, username)#

Updates the owner of a book and records the ownership history.

Parameters:

  • id (str) – The unique identifier of the book.

  • username (str) – The username of the new owner.

Returns:

The updated book data including the new owner and ownership history.

Return type:

dict

application.db.book.share_book_with_non_user(book_id, name)#

Share a book with a non-user by updating the book’s share history.

Parameters:

  • book_id (str) – The ID of the book to be shared.

  • name (str) – The name of the non-user with whom the book is being shared.

Returns:

The book data after the share operation.

Return type:

dict

Raises:

exceptions.BookTagDoesNotExistError – If the book with the given ID does not exist.

application.db.book.share_book_with_user(book_id, username)#

Marke a book as shared (with a user), updating the book’s share history.

Parameters:

  • book_id (str) – The ID of the book to be shared.

  • username (str) – The username of the user with whom the book is being shared.

Returns:

The book data after the share operation.

Return type:

dict

Raises:

exceptions.BookTagDoesNotExistError – If the book with the given ID does not exist.

application.db.book.subsonic_init()#
Return type:

SubsonicClient

Unlink a book tag from the database using its RFID.

This function searches for a book in the database using the provided RFID. If the book is found, it deletes the book entry from the database and removes any reference to its thumbnail if it is a locally hosted blob.

Parameters:

rfid (str) – The RFID of the book to be unlinked.

Returns:

The data of the book that was unlinked.

Return type:

dict

Raises:

BookTagDoesNotExistError – If no book with the given RFID is found in the database.

application.db.bugs module#

This module handles all bug report functionalities, including reporting new bugs, commenting on bugs, retrieving bug reports, and managing their statuses.

application.db.bugs.comment_on_bug(id, text, plaintext=True)#

Add a comment to a bug report.

Parameters:

  • id (str) – The unique identifier of the bug report.

  • text (str) – The content of the comment.

  • plaintext (bool, optional) – If True, the comment is treated as plain text. If False, the comment is treated as Markdown. Defaults to True.

Returns:

The updated bug report.

Return type:

dict

Raises:

BugReportDoesNotExistError – If the bug report with the given id does not exist.

application.db.bugs.count_bug_reports(userids, resolved)#

Count the number of bug reports based on the given user IDs and resolution status.

Parameters:

  • userids (list) – A list of user IDs to filter the bug reports. If the list is empty, the count will be based only on the resolution status.

  • resolved (bool) – The resolution status to filter the bug reports.

Returns:

The count of bug reports that match the given criteria.

Return type:

int

application.db.bugs.delete_bug_report(id)#

Deletes a bug report from the database by its ID.

Parameters:

id (str) – The ID of the bug report to delete.

Returns:

The deleted bug report with its ID included.

Return type:

dict

Raises:

BugReportDoesNotExistError – If no bug report with the given ID exists.

application.db.bugs.get_bug_report(id)#

Retrieve a bug report from the database by its ID.

Parameters:

id (str) – The unique identifier of the bug report.

Returns:

The processed bug report data.

Return type:

dict

Raises:

BugReportDoesNotExistError – If no bug report with the given ID is found.

application.db.bugs.get_bug_reports(userids, start, count, resolved)#

Retrieve a list of bug reports from the database.

Parameters:

  • userids (list) – A list of user IDs to filter the bug reports by creator.

  • start (int) – The starting index for pagination.

  • count (int) – The number of bug reports to retrieve.

  • resolved (bool) – A flag indicating whether to retrieve resolved or unresolved bug reports.

Returns:

A list of processed bug reports.

Return type:

list

application.db.bugs.process_bug_report(report)#

Prepares a bug report database item for display by populating the report’s creator and comments with usernames.

Parameters:

report (dict) – A dictionary containing the bug report details. Expected keys are ‘_id’, ‘creator’, and ‘convo’. ‘convo’ is a list of comments, each with a ‘creator’ key.

Returns:

The processed bug report with updated ‘id’, ‘creator’, and ‘convo’ fields.

The ‘id’ field is set to the value of ‘_id’. The ‘creator’ field is updated to the username if the user exists, otherwise it is converted to a string. Each comment’s ‘creator’ field is similarly updated.

Return type:

dict

application.db.bugs.report_bug(text, plaintext=True)#

Report a new bug to the database.

Parameters:

  • text (str) – The content of the bug report.

  • plaintext (bool, optional) – If True, the text is treated as plain text and escaped for HTML. If False, the text is treated as Markdown and converted to HTML. Defaults to True.

Returns:

A dictionary containing the bug report details, including the newly assigned bug report ID.

Return type:

dict

application.db.bugs.set_bug_status(id, status)#

Update the status of a bug report and notify the creator.

Parameters:

  • id (str) – The unique identifier of the bug report.

  • status (bool) – The new status of the bug report. True for resolved, False for reopened.

Returns:

The updated bug report.

Return type:

dict

Raises:

exceptions.BugReportDoesNotExistError – If the bug report with the given id does not exist.

application.db.datafeed module#

application.db.datafeed

application.db.datafeed.count_documents(feed)#

Count the number of documents in the database for a given feed.

Parameters:

feed (str) – The feed identifier, expected to be a valid ObjectId.

Returns:

The number of documents associated with the given feed.

Returns 0 if the feed identifier is not valid.

Return type:

int

application.db.datafeed.count_feeds()#

Count the number of documents in the ‘feeds’ collection.

Returns:

The total number of documents in the ‘feeds’ collection.

Return type:

int

application.db.datafeed.create_document(feed, author, posted, body, title, url)#

Create a new document in the database.

Parameters:

  • feed (str) – The feed identifier.

  • author (str | None) – The author of the document. Can be None.

  • posted (datetime | None) – The date and time the document was posted. Can be None.

  • body (str) – The body content of the document.

  • title (str | None) – The title of the document. Can be None.

  • url (str) – The URL of the document.

Returns:

The processed document from the database.

Return type:

dict

application.db.datafeed.create_feed(name, url, kind, notify)#

Creates a new feed in the database.

Parameters:

  • name (str) – The name of the feed.

  • url (str) – The URL of the feed.

  • kind (str) – The kind of the feed. Must be ‘markdown_recursive’.

  • notify (bool) – Whether to notify users about the feed.

Returns:

The created feed data.

Return type:

dict

Raises:

InvalidFeedKindError – If the kind is not ‘markdown_recursive’.

application.db.datafeed.delete_feed(id)#

Deletes a feed and its associated documents from the database.

Parameters:

id (str) – The unique identifier of the feed to be deleted.

Returns:

A dictionary representation of the deleted feed after processing.

Return type:

dict

Raises:

FeedDoesNotExistError – If the feed with the given id does not exist.

application.db.datafeed.get_body_html(feed_kind, body)#

Converts the given body text to HTML based on the specified feed kind.

Parameters:

  • feed_kind (str) – The type of feed to process. Currently supports ‘markdown_recursive’.

  • body (str) – The body text to be converted to HTML.

Returns:

The converted HTML string.

Return type:

str

Raises:

InvalidFeedKindError – If the provided feed kind is not supported.

application.db.datafeed.get_document(id)#

Retrieve a document from the database by its ID.

Parameters:

id (str) – The ID of the document to retrieve.

Returns:

The processed document.

Return type:

dict

Raises:

FeedDocumentDoesNotExistError – If the document does not exist or the ID is invalid.

application.db.datafeed.get_documents(feed, start, count, sorting)#

Retrieve a list of documents from the database based on the specified feed, starting index, count, and sorting order.

Parameters:

  • feed (str) – The ID of the feed to retrieve documents from.

  • start (int) – The starting index of the documents to retrieve.

  • count (int) – The number of documents to retrieve.

  • sorting (Sorting) – The sorting criteria for the documents. It should be a dictionary with ‘fields’ as a list of field names and ‘descending’ as a boolean indicating the sort order.

Returns:

A list of documents matching the specified criteria.

Each document is represented as a dictionary.

Return type:

list[dict]

application.db.datafeed.get_feed(id)#

Retrieve a feed from the database by its ID.

Parameters:

id (str) – The ID of the feed to retrieve.

Returns:

The processed feed data.

Return type:

dict

Raises:

FeedDoesNotExistError – If the provided ID is not valid or if no feed is found with the given ID.

application.db.datafeed.get_feeds(start, count)#

Retrieve a list of processed feeds from the database.

Parameters:

  • start (int) – The starting index from which to retrieve feeds.

  • count (int) – The number of feeds to retrieve.

Returns:

A list of dictionaries, each representing a processed feed.

Return type:

list[dict]

application.db.datafeed.get_user_feeds(username)#

Retrieve the feeds created by a specific user.

Parameters:

username (str) – The username of the user whose feeds are to be retrieved.

Returns:

A list of dictionaries, each representing a feed created by the user.

Return type:

list[dict]

application.db.datafeed.process_document(document)#

Processes a document by copying the value of the ‘_id’ key to a new ‘id’ key.

Parameters:

document (dict) – The document to be processed.

Returns:

The processed document with the ‘id’ key added.

Return type:

dict

application.db.datafeed.process_feed(feed)#

Processes a feed dictionary by updating its ‘id’ and ‘creator’ fields.

Parameters:

feed (dict) – The feed dictionary to be processed. It must contain ‘_id’ and ‘creator’ keys.

Returns:

The processed feed dictionary with ‘id’ and ‘creator’ fields updated.

Return type:

dict

Raises:

UserDoesNotExistError – If the user with the given ‘creator’ ID does not exist.

application.db.datafeed.set_document_read(id, read)#

Sets the ‘read’ status of a document in the database.

Parameters:

  • id (str) – The unique identifier of the document.

  • read (bool) – The read status to set for the document.

Returns:

The updated document with the new ‘read’ status.

Return type:

dict

application.db.datafeed.set_feed_inactive(id, inactive)#

Sets the ‘inactive’ status of a feed in the database.

If the feed is set to inactive, new documents will not be fetched and added to the feed.

Parameters:

  • id (str) – The unique identifier of the feed.

  • inactive (bool) – The status to set for the feed’s ‘inactive’ field.

Returns:

The updated feed document with the new ‘inactive’ status.

Return type:

dict

application.db.datafeed.set_feed_navigation(id, page, sorting)#

Updates the navigation settings for a specific feed in the database.

Parameters:

  • id (str) – The unique identifier of the feed.

  • page (int | None) – The current page number to set for the feed. Can be None.

  • sorting (Sorting | None) – The sorting criteria to set for the feed. Can be None.

Returns:

The updated feed with the new navigation settings.

Return type:

dict

application.db.datafeed.set_feed_notify(id, notify)#

Updates the notification setting for a specific feed and returns the updated feed data.

Parameters:

  • id (str) – The unique identifier of the feed.

  • notify (bool) – The new notification setting for the feed.

Returns:

The updated feed data with the new notification setting.

Return type:

dict

application.db.datafeed.update_document(id, body)#

Update the body of a document in the database.

This function retrieves a document by its ID, updates its body and body_html fields, and sets the updated timestamp to the current UTC time. The updated document is then saved back to the database.

Parameters:

  • id (str) – The ID of the document to update.

  • body (str) – The new body content for the document.

Returns:

The updated document.

Return type:

dict

application.db.documents module#

Allows users to create, read and edit arbitrary rich text documents.

application.db.documents.create_document(title, body, parent=None)#

Creates a new document in the database.

Parameters:

  • title (str) – The title of the document.

  • body (str) – The content of the document.

  • parent (str | None, optional) – The ID of the parent document. Defaults to None.

Returns:

The new document.

Return type:

dict

application.db.documents.delete_document(doc_id)#

Deletes a document from the database.

Parameters:

doc_id (str) – The ID of the document to delete.

Returns:

The deleted document.

Return type:

dict

application.db.documents.get_child_documents(doc_id)#

Retrieves the child documents of a document.

Parameters:

doc_id (str | None) – The optional ID of the document. If None, retrieves top-level documents.

Returns:

The child documents.

Return type:

list

application.db.documents.get_document(doc_id)#

Retrieves a document from the database by its ID.

Parameters:

doc_id (str) – The ID of the document to retrieve.

Returns:

The document.

Return type:

dict

application.db.documents.parse_document(doc)#

Parses a document from the database.

Parameters:

doc (dict) – The document to parse.

Returns:

The parsed document.

Return type:

dict

application.db.documents.update_document(doc_id, title, body, new_parent)#

Updates a document in the database.

Parameters:

  • doc_id (str) – The ID of the document to update.

  • title (str | None) – The new title of the document. If None, the title is not updated.

  • body (str | None) – The new content of the document. If None, the body is not updated.

  • new_parent (str | None) – The ID of the new parent document. If None, the parent is not updated.

Returns:

The updated document.

Return type:

dict

application.db.inventory module#

application.db.inventory

application.db.inventory.build_inventory_query(filter)#

Builds a MongoDB query dictionary for searching inventory based on the provided filter and user ID.

Parameters:

  • filter (InventorySearchFilter) – An object containing search criteria for the inventory.

  • user_id (ObjectId) – The ID of the user making the query.

Returns:

A MongoDB query dictionary constructed based on the provided filter criteria.

Return type:

dict

application.db.inventory.count_inventory(filter)#

Count the number of inventory items based on the provided filter and user ID.

Parameters:

  • filter (InventorySearchFilter) – The filter criteria to apply to the inventory search.

  • user_id (ObjectId) – The ID of the user whose inventory is being queried.

Returns:

The number of inventory items that match the filter criteria for the specified user.

Returns 0 if the user does not exist.

Return type:

int

Raises:

exceptions.UserDoesNotExistError – If the user does not exist.

application.db.inventory.create_inventory_item(owner, category, type, location, blob_id, description, rfid)#

Create a new inventory item in the database.

Parameters:

  • owner (str) – The username of the owner of the item.

  • category (str) – The category of the item.

  • type (str) – The type of the item.

  • location (str) – The location of the item.

  • blob_id (str) – The ID of the associated blob.

  • description (str) – A textual description of the item.

  • rfid (str | None) – The RFID tag of the item, if any.

Returns:

The created inventory item.

Return type:

dict

Raises:

exceptions.ItemExistsError – If an item with the given RFID already exists.

application.db.inventory.delete_inventory_item(id)#

Deletes an inventory item from the database by its ID.

Parameters:

id (str) – The ID of the inventory item to be deleted.

Returns:

The deleted inventory item details.

Return type:

dict

Raises:

ValueError – If the inventory item with the given ID does not exist.

application.db.inventory.get_inventory(filter, start, count, sorting)#

Retrieves a list of inventory items based on the provided filter, pagination, and sorting options.

Parameters:

  • filter (InventorySearchFilter) – The filter criteria to apply to the inventory search.

  • start (int) – The starting index for pagination.

  • count (int) – The number of items to retrieve.

  • sorting (Sorting) – The sorting criteria for the inventory items.

  • user_id (ObjectId) – The ID of the user making the request.

Returns:

A list of inventory items matching the search criteria.

Return type:

list

Raises:

exceptions.UserDoesNotExistError – If the user specified in the filter does not exist.

application.db.inventory.get_inventory_item(id)#

Retrieve an inventory item from the database by its ID.

Parameters:

id (str) – The ID of the inventory item to retrieve.

Returns:

A dictionary representing the inventory item.

Return type:

dict

Raises:

exceptions.ItemDoesNotExistError – If no item with the given ID is found.

application.db.inventory.get_item_categories()#

Retrieve a list of distinct item categories from the database.

Returns:

A list of unique item categories.

Return type:

list[str]

application.db.inventory.get_item_locations(owner)#

Retrieve a list of distinct item locations for a given owner.

Parameters:

owner (str) – The username of the owner whose item locations are to be retrieved.

Returns:

A list of distinct locations where the owner’s items are stored.

Return type:

list[str]

application.db.inventory.get_item_types(category)#

Retrieve a list of distinct item types for a given category from the database.

Parameters:

category (str) – The category of items to filter by.

Returns:

A list of distinct item types within the specified category.

Return type:

list[str]

Change the RFID of an inventory item to a new one.

Parameters:

  • id (str) – The ID of the inventory item.

  • rfid (str | None) – The new RFID for the inventory item, or None to unlink

Returns:

The updated inventory item.

Return type:

dict

Raises:

application.db.notification module#

This module allows sending web push notifications to users.

application.db.notification.count_notifications(username, read)#

Count the number of notifications for a given user based on their read status.

Parameters:

  • username (str) – The username of the user whose notifications are to be counted.

  • read (bool) – The read status of the notifications to be counted (True for read, False for unread).

Returns:

The number of notifications that match the given criteria.

Return type:

int

application.db.notification.create_subscription(username, subscription_token)#

Create a notification subscription for a user.

Parameters:

  • username (str) – The username of the user.

  • subscription_token (dict) – A dictionary containing the subscription token details.

  • are (Expected keys) –

    • ‘endpoint’ (str): The endpoint URL of the subscription.

    • ’expirationTime’ (str | None): The expiration time of the subscription.

    • ’keys’ (dict): A dictionary containing the keys for the subscription.
      Expected keys are:

      • ’p256dh’ (str): The p256dh key.

      • ’auth’ (str): The auth key.

Raises:

exceptions.InvalidSubscriptionToken – If the subscription token is invalid.

Return type:

None

application.db.notification.delete_subscription(auth)#

Deletes all notification subscriptions from the database that match the given authentication token.

Parameters:

auth (str) – The authentication token used to identify subscriptions to delete.

Returns:

The number of subscriptions deleted.

Return type:

int

application.db.notification.delete_subscriptions(username)#

Delete all notification subscription records for a given user.

Parameters:

username (str) – The username of the user whose subscriptions are to be deleted.

Returns:

The number of subscription records that were deleted.

Return type:

int

application.db.notification.get_notifications(username, read, start, count)#

Retrieve a list of notifications for a given user.

Parameters:

  • username (str) – The username of the recipient.

  • read (bool) – Filter notifications based on their read status.

  • start (int) – The starting index for pagination.

  • count (int) – The number of notifications to retrieve.

Returns:

A list of notifications, each represented as a dictionary.

Return type:

list

application.db.notification.get_public_key()#

Retrieve the VAPID public key.

Returns:

The VAPID public key as a string.

Return type:

str

application.db.notification.get_subscription(auth)#

Retrieve a notification subscription token from the database using the provided authentication key.

Parameters:

auth (str) – The authentication key used to find the subscription.

Returns:

The subscription token if found, otherwise None.

Return type:

dict | None

application.db.notification.get_subscriptions(username)#

Retrieve a list of notification subscription tokens for a given user.

Parameters:

username (str) – The username of the user whose subscriptions are to be retrieved.

Returns:

A list of subscription tokens associated with the user.

Return type:

list

application.db.notification.get_user_from_notif(id)#

Retrieve user information based on a notification ID.

Parameters:

id (str) – The ID of the notification.

Returns:

A dictionary containing user information if found, otherwise an empty dictionary.

Return type:

dict

application.db.notification.init()#

Initialize the notification module by loading VAPID keys from the database, or generating them if they do not exist.

Return type:

None

application.db.notification.mark_all_as_read(username)#

Marks all notifications as read for a given user.

Parameters:

username (str) – The username of the user whose notifications are to be marked as read.

Return type:

None

Returns:

None

application.db.notification.mark_as_read(item_id)#

Marks a notification as read in the database.

Parameters:

item_id (str) – The unique identifier of the notification to be marked as read.

Return type:

None

Returns:

None

application.db.notification.send(title, body, username, *, category='general', read=False)#

Send a notification to a user and log the notification in the database.

Parameters:

  • title (str) – The title of the notification.

  • body (str) – The body content of the notification.

  • username (str) – The username of the recipient.

  • category (str, optional) – The category of the notification. Defaults to ‘general’.

  • read (bool, optional) – Whether the notification is marked as read. Defaults to False.

Returns:

A dictionary containing the logged notification data.

Return type:

dict

Raises:

exceptions.MissingConfig – If the admin email configuration is missing.

application.db.notification.try_send_webpush(username, subscription_token, message, admin_email, endpoint)#

Attempts to send a web push notification to a user and handles exceptions.

Parameters:

  • username (str) – The username of the notification recipient.

  • subscription_token (dict) – The web push subscription information for the user.

  • message (dict) – The notification message payload to send.

  • admin_email (str) – The administrator’s email address for VAPID claims.

  • endpoint (str) – The endpoint URL for the push service.

Returns:

True if the notification was sent successfully, False otherwise.

Return type:

bool

application.db.perms module#

This module provides logic for checking user permissions and ownership of data. Permissions are a big part of the application, allowing users to only perform actions they are allowed to, based on their roles and the data they are interacting with.

Likewise, it provides a way to check if the user has access to specific modules (features) within the application. Unlike permissions which are per-user, modules can be managed server-wide, per group, or per user.

Resolvers can use the require and require_all decorators to enforce permissions on the calling user before executing the resolver function, and the module decorator to check if the user has access to specific modules.

application.db.perms.bad_perms()#

Returns a dictionary representing an insufficient permissions error.

Returns:

A dictionary containing the error type and message indicating

that the user is not allowed to perform the requested action.

Return type:

dict[str, str]

application.db.perms.caller_info()#

Retrieves caller information based on the request token.

The function attempts to decode the request token to extract the username. If decoding fails, it treats the token as an API key and retrieves the corresponding API key information from the database.

Returns:

The user data associated with the username if available,

otherwise the API key information if the token is an API key. Returns None if neither is found or an error occurs during retrieval.

Return type:

dict[str, Any] | None

application.db.perms.caller_info_strict()#

Retrieves caller information based on the request token.

The function attempts to decode the request token to extract the username. If decoding fails, it treats the token as an API key and retrieves the corresponding API key information from the database.

Returns:

The user data associated with the username if available,

otherwise the API key information if the token is an API key.

Return type:

UserData | dict

Raises:
application.db.perms.module(*modules)#

Require the calling user to have all the specified modules enabled.

This is a decorator for application resolvers, to avoid redundant module-checking logic all over the place. If the user does not have one or more of the specified modules enabled when the resolver is called, then the resolver will be overridden and will instead return a bad_perms() dict.

Parameters:

modules (list[str]) – The modules that must ALL be enabled for the user.

Returns:

The resolver function, with decorator applied.

Return type:

Callable

application.db.perms.require(*perms, perform_on_self=False, data_func=None)#

Require the calling user to have any of the specified permissions.

This is a decorator for application resolvers, to avoid redundant permission-checking logic all over the place. If the permissions are not satisfied when the resolver is called, then the resolver will be overridden and will instead return a bad_perms() dict.

Parameters:

  • *perms (str) – The permissions that must have at least 1 satisfied.

  • perform_on_self (bool | tuple[bool, ...]) – If True, permissions will be ignored when the user is editing their own data.

  • data_func (Callable|None) – If specified, this function will give the data to be checked for ownership. Otherwise, the main function’s parameters are checked.

Returns:

The resolver function, with decorator applied.

Return type:

Callable

application.db.perms.require_all(*perms, perform_on_self=False, data_func=None)#

Require the calling user to have all of the specified permissions.

This is a decorator for application resolvers, to avoid redundant permission-checking logic all over the place. If the permissions are not satisfied when the resolver is called, then the resolver will be overridden and will instead return a bad_perms() dict.

Parameters:

  • *perms (str) – The permissions that must all be satisfied.

  • perform_on_self (bool | tuple[bool, ...]) – If True, permissions will be ignored when the user is editing their own data.

  • data_func (Callable|None) – If specified, this function will give the data to be checked for ownership. Otherwise, the main function’s parameters are checked.

Returns:

The resolver function, with decorator applied.

Return type:

Callable

application.db.perms.satisfies(perms, data=None, *, perform_on_self=False, data_func=None)#

Check if the calling user has any of the given permissions.

Parameters:

  • perms (list[str]) – The permissions that must have at least 1 satisfied.

  • perform_on_self (bool | tuple[bool, ...]) – If True, permissions will be ignored when the user is editing their own data.

  • data_func (Callable|None) – If specified, this function will give the data to be checked for ownership. Otherwise, the main function’s parameters are checked.

Returns:

True if the user has all required permissions, or if perform_on_self is True

AND the data being operated on belongs to the user. If the user does not have the required permissions, returns a dictionary representing an insufficient permissions error. Otherwise, returns False.

Return type:

bool | dict

application.db.perms.user_has_perms(user_data, perm_list)#

Check if the user has any of the specified permissions.

Parameters:

  • user_data (UserData | dict) – A dictionary containing user information, including a ‘perms’ key with a list of permissions.

  • perm_list (list) – A list of permissions to check against the user’s permissions.

Returns:

True if the user has any of the specified permissions, False otherwise.

Return type:

bool

application.db.sessions module#

This module provides user session management functionalities.

A session is created when a user logs in, and a valid session token is required for all API requests aside from logging in.

application.db.sessions.count_valid_sessions(username)#

Count the number of valid sessions for a given username.

Parameters:

username (str) – The username to count sessions for.

Returns:

The number of valid sessions associated with the username.

Return type:

int

application.db.sessions.get_first_session_token(username)#

Retrieve the first session token for a given username.

Parameters:

username (str) – The username to search for in the database.

Returns:

The session token if found, otherwise None.

Return type:

str | None

application.db.sessions.revoke_sessions(username)#

Revoke all sessions for a given username.

This function deletes all session records associated with the specified username from the database.

Parameters:

username (str) – The username whose sessions are to be revoked.

Return type:

None

Returns:

None

application.db.sessions.start_session(token, username)#

Starts a new session for the given user.

This function retrieves user data for the specified username and creates a new session in the database with the provided token. The session expiry time is determined based on the user’s permissions. Persistent logins last for 20 weeks, while non-persistent logins last for 7 days.

Parameters:

  • token (str) – The session token.

  • username (str) – The username of the user starting the session.

Return type:

None

Returns:

None

application.db.sessions.valid_session(token)#

Check if a session with the given token exists in the database.

Parameters:

token (str) – The session token to validate.

Returns:

True if a session with the token exists, False otherwise.

Return type:

bool

application.db.settings module#

application.db.settings

application.db.settings.calculate_disabled_modules(disabled_modules)#

Calculate the complete list of disabled modules based on the provided list of disabled modules and the module configuration.

Parameters:

disabled_modules (list[str]) – A list of module IDs that are initially disabled.

Returns:

A list of module IDs that are disabled, including those that are

indirectly disabled due to dependencies.

Return type:

list[str]

application.db.settings.create_theme(theme)#

Creates or updates a theme in the database.

If a theme with the same name already exists, it updates the existing theme. Otherwise, it inserts a new theme.

Parameters:

theme (dict) – A dictionary containing the theme details.

Returns:

The theme that was created or updated.

Return type:

dict

application.db.settings.delete_theme(name)#

Deletes a theme from the database by its name.

Parameters:

name (str) – The name of the theme to delete.

Returns:

The deleted theme document.

Return type:

dict

Raises:

exceptions.MissingConfig – If the theme with the specified name does not exist.

application.db.settings.get_all_configs()#

Retrieve all configuration entries from the database.

Returns:

A list of dictionaries, each containing ‘name’ and ‘value’ keys

representing the configuration entries.

Return type:

list

application.db.settings.get_all_themes()#

Retrieve all themes from the database.

Returns:

A list of all themes found in the database.

Return type:

list

application.db.settings.get_config(name)#

Retrieve the configuration value for a given configuration name.

Parameters:

name (str) – The name of the configuration to retrieve.

Returns:

The value of the configuration if found, otherwise None.

Return type:

str | None

application.db.settings.get_enabled_modules(user_data=None, *, group=None)#

Retrieve a list of enabled modules, optionally filtering out those disabled for a specific user or group.

Parameters:

  • user_data (UserData | dict | None, optional) – A dictionary containing user-specific data, including disabled modules and groups. Defaults to None.

  • group (str | None, optional) – A specific group to consider for disabled modules. Defaults to None.

Returns:

A list of enabled modules, excluding those disabled for the specified user or group.

Return type:

list

application.db.settings.get_groups()#

Retrieve a list of user groups from the database.

Returns:

A list of groups if the ‘groups’ document is found, otherwise an empty list.

Return type:

list

application.db.settings.get_modules(user_data)#

Retrieve a list of modules available to a user, excluding any disabled modules based on the user’s groups.

Parameters:

user_data (UserData | dict) – A dictionary containing user information, including their groups.

Returns:

A list of enabled modules for the user, excluding any disabled modules.

Return type:

list

application.db.settings.set_config(name, value)#

Sets or deletes a configuration value in the database.

If the value is None, the configuration entry with the given name is deleted. Otherwise, the configuration entry is updated or created with the provided value.

Parameters:

  • name (str) – The name of the configuration setting.

  • value (str | None) – The value to set for the configuration setting. If None, the setting is deleted.

Return type:

None

Returns:

None

application.db.settings.set_module_enabled(module_id, enabled, group)#

Enable or disable a module in the database.

If a group is specified, the module’s enabled/disabled state is updated within that group. Otherwise, the module’s state is updated globally.

Parameters:

  • module_id (str) – The ID of the module to be enabled or disabled.

  • enabled (bool) – True to enable the module, False to disable it.

  • group (str | None) – The group within which to update the module’s state. If None, the global state is updated.

Return type:

None

Returns:

None

application.db.settings.update_groups(old_groups, new_groups)#

Updates the user counts for groups in the database based on the provided old and new group lists.

If the ‘groups’ document does not exist in the database, it creates a new one with the new groups.

Parameters:

  • old_groups (list[str]) – A list of group names that the user was previously a member of.

  • new_groups (list[str]) – A list of group names that the user is currently a member of.

Return type:

None

Returns:

None

application.db.users module#

application.db.users

application.db.users.authenticate(username, password)#

Authenticates the user with the specified username and password.

Parameters:

  • username (str) – The username of the user.

  • password (str) – The password of the user.

Returns:

A login token if authentication is successful.

Return type:

str

Raises:
application.db.users.count_users()#
Returns:

The count of all non-ephemeral users.

Return type:

int

application.db.users.create_user(username, password, *, groups, admin=False, ephemeral=False)#

Creates a new user with the specified username, password, groups, admin status, and ephemeral status.

Parameters:

  • username (str) – The username of the new user.

  • password (str) – The password of the new user.

  • groups (list, optional) – List of groups the user belongs to. Defaults to [].

  • admin (bool, optional) – Whether the user is an admin. Defaults to False.

  • ephemeral (bool, optional) – Whether the user is ephemeral. Defaults to False.

Returns:

The created user data.

Return type:

dict

Raises:
application.db.users.delete_user(username)#

Deletes the user with the specified username.

Parameters:

username (str) – The username of the user to delete.

Returns:

The deleted user data.

Return type:

dict

Raises:

UserDoesNotExistError – If the user is not found.

application.db.users.dump_collection(collection, queries, fp)#

Exports documents from a specified MongoDB collection based on provided queries and writes them to a ZipFile as JSON. Any sensitive information, such as passwords, is excluded from the export.

Parameters:

  • collection (str) – The name of the MongoDB collection to export.

  • queries (list) – A list of query dictionaries to filter documents for export.

  • fp (ZipFile) – An open ZipFile object to which the exported JSON will be written.

Return type:

None

application.db.users.export_user_data(username)#

Exports the user data for the user with the specified username.

Parameters:

username (str) – The username of the user.

Returns:

A blob containing the exported data.

Return type:

dict

Raises:

UserDoesNotExistError – If the user is not found.

application.db.users.get_admins()#
Returns:

A list of all admin users.

Return type:

list

application.db.users.get_user_by_id(id)#

Returns user data for the user with the specified ID.

Parameters:

id (ObjectId) – The ID of the user.

Returns:

The user data.

Return type:

UserData

Raises:

UserDoesNotExistError – If the user is not found.

application.db.users.get_user_data(username)#

Returns user data for the user with the specified username.

Parameters:

username (str) – The username of the user.

Returns:

The user data.

Return type:

dict

Raises:

UserDoesNotExistError – If the user is not found.

application.db.users.get_user_list(groups)#

Returns a list of users with their username, display name, and last login.

Parameters:

groups (list, optional) – List of groups to filter users by. Defaults to [].

Returns:

A list of users.

Return type:

list

application.db.users.group_filter(filter, user_data)#

Applies group filtering to the specified filter based on the user’s groups.

Parameters:

  • filter (Filter) – The filter to apply group filtering to.

  • user_data (dict) – The user data containing the groups.

Returns:

The updated filter.

Return type:

dict

application.db.users.is_locked(failed_logins)#

Checks if a user is locked based on the number of failed login attempts.

Parameters:

failed_logins (int) – The number of failed login attempts.

Returns:

True if the user is locked, False otherwise.

Return type:

bool

application.db.users.login_attempts_remaining(failed_logins)#

Calculates the number of login attempts remaining before the user is locked.

Parameters:

failed_logins (int) – The number of failed login attempts.

Returns:

The number of login attempts remaining.

Return type:

int

application.db.users.process_user_data(userdata)#

Process user data to ensure it has the correct structure.

Parameters:

userdata (dict) – The user data to process.

Returns:

The processed user data.

Return type:

UserData

application.db.users.unlock_user(username)#

Unlocks the user with the specified username by resetting their failed login attempts.

Parameters:

username (str) – The username of the user to unlock.

Returns:

The updated user data.

Return type:

UserData

Raises:

UserDoesNotExistError – If the user is not found.

application.db.users.update_user_display_name(username, display_name)#

Updates the display name for the user with the specified username.

Parameters:

  • username (str) – The username of the user.

  • display_name (str) – The new display name.

Returns:

The updated user data.

Return type:

dict

Raises:
application.db.users.update_user_email(username, email)#

Updates the email for the user with the specified username.

Parameters:

  • username (str) – The username of the user.

  • email (str) – The new email.

Returns:

The updated user data.

Return type:

dict

Raises:
application.db.users.update_user_groups(username, groups)#

Updates the groups for the user with the specified username.

Parameters:

  • username (str) – The username of the user.

  • groups (list) – The new groups.

Returns:

The updated user data.

Return type:

dict

Raises:
application.db.users.update_user_module(username, module, disabled)#

Updates the disabled modules for the user with the specified username.

Parameters:

  • username (str) – The username of the user.

  • module (str) – The module to update.

  • disabled (bool) – Whether the module is disabled.

Returns:

The updated user data.

Return type:

dict

Raises:
application.db.users.update_user_password(username, password)#

Updates the password for the user with the specified username.

Parameters:

  • username (str) – The username of the user.

  • password (str) – The new password.

Returns:

The updated user data.

Return type:

dict

Raises:
application.db.users.update_user_perms(username, perms)#

Updates the permissions for the user with the specified username.

Parameters:

  • username (str) – The username of the user.

  • perms (list) – The new permissions.

Returns:

The updated user data.

Return type:

dict

Raises:

UserDoesNotExistError – If the user is not found.

application.db.users.update_user_theme(username, theme)#

Updates the theme for the user with the specified username.

Parameters:

  • username (str) – The username of the user.

  • theme (dict) – The new theme data.

Returns:

The updated user data.

Return type:

dict

Raises:

UserDoesNotExistError – If the user is not found.

application.db.users.update_username(username, new_username)#

Updates the username for the user with the specified username.

Parameters:

  • username (str) – The current username of the user.

  • new_username (str) – The new username.

Returns:

The updated user data.

Return type:

dict

Raises:
application.db.users.userids_in_groups(groups)#

Returns a list of user IDs for users in the specified groups.

Parameters:

groups (list) – List of groups to filter users by.

Returns:

A list of user IDs.

Return type:

list

application.db.weather module#

application.db.weather

application.db.weather.count_alert_history(username)#

Count the number of alert history documents in the database.

Parameters:

username (str | None) – The username to filter the alert history by. If None, count all alert history documents.

Returns:

The count of alert history documents matching the criteria.

Return type:

list

application.db.weather.create_user(user_data)#

Creates a new weather user in the database.

This function first checks if the user exists in the main users collection. If the user does not exist, it raises a UserDoesNotExistError. If the user already exists in the weather_users collection, it raises a UserExistsError.

The function then processes the ‘max’ and ‘min’ fields in the user_data to determine their values based on the ‘disable’ and ‘default’ flags.

Finally, it inserts the new user data into the weather_users collection and returns the processed weather user data.

Parameters:

user_data (dict) – A dictionary containing user information. Expected keys are: - ‘username’ (str): The username of the user. - ‘lat’ (float): The latitude of the user’s location. - ‘lon’ (float): The longitude of the user’s location. - ‘max’ (dict): A dictionary with ‘disable’ (bool) and ‘default’ (bool) keys. - ‘min’ (dict): A dictionary with ‘disable’ (bool) and ‘default’ (bool) keys.

Returns:

The processed weather user data.

Return type:

dict

Raises:
application.db.weather.delete_user(username)#

Delete a user from the weather database.

Parameters:

username (str) – The username of the user to be deleted.

Returns:

The user’s previous (now nonexistent) weather information.

Return type:

dict

application.db.weather.get_alert_history(username, start, count)#

Retrieve a list of alert history records from the database.

Parameters:

  • username (str | None) – The username to filter the alert history by. If None, retrieves all alert history.

  • start (int) – The starting index for the records to retrieve.

  • count (int) – The number of records to retrieve.

Returns:

A list of dictionaries containing alert history records. Each dictionary contains:

  • ’recipient’ (str): The recipient of the alert.

  • ’message’ (str): The message content of the alert.

  • ’sent’ (datetime): The timestamp when the alert was sent.

Return type:

list

application.db.weather.get_last_exec()#

Retrieves the most recent weather log entry from the database.

Returns:

The most recent weather log entry as a dictionary if it exists,

otherwise None.

Return type:

dict | None

application.db.weather.get_users()#

Retrieve and process weather users from the database.

This function fetches all weather users from the db.weather_users collection, processes each user using the process_weather_user function, and returns a sorted list of processed users. The sorting is based on the concatenation of the integer value of the ‘exclude’ field and the ‘username’ field.

Returns:

A sorted list of processed weather users.

Return type:

list

application.db.weather.get_weather_user(username)#

Retrieve weather user data from the database based on the provided username.

Parameters:

username (str) – The username of the user whose weather data is to be retrieved.

Returns:

A dictionary containing the weather user data.

Return type:

dict

Raises:

UserDoesNotExistError – If the user with the given username does not exist in the database.

application.db.weather.log_user_weather_alert(username, message)#

Logs a weather alert message for a user.

This function updates the last sent time for the user’s weather alert and logs the alert message in the alert history.

Parameters:

  • username (str) – The username of the user to log the alert for.

  • message (str) – The weather alert message to log.

Return type:

None

Returns:

None

application.db.weather.log_weather_alert(users, error)#

Logs a weather alert to the database.

Parameters:

  • users (list[str]) – A list of user identifiers who are affected by the weather alert.

  • error (str | None) – An optional error message if there was an issue with the weather alert.

Return type:

None

Returns:

None

application.db.weather.process_weather_user(userdata)#

Processes the weather-related user data.

This function takes a dictionary containing user data, validates the user ID, retrieves additional user information from the database if necessary, and processes the ‘max’ and ‘min’ weather values.

Parameters:

userdata (dict) – A dictionary containing user data. Expected keys are: - ‘_id’: The user ID, which should be a valid ObjectId. - ‘max’: The maximum weather value (optional, should be a float). - ‘min’: The minimum weather value (optional, should be a float).

Returns:

The processed user data dictionary with the following keys:

  • ’username’: The username associated with the user ID, or the user ID itself if not found.

  • ’max’: A dictionary with the following keys:

    • ’value’: The maximum weather value (float).

    • ’default’: A boolean indicating if the ‘max’ value was not provided.

    • ’disable’: A boolean indicating if the ‘max’ value is explicitly set to False.

  • ’min’: A dictionary with the following keys:

    • ’value’: The minimum weather value (float).

    • ’default’: A boolean indicating if the ‘min’ value was not provided.

    • ’disable’: A boolean indicating if the ‘min’ value is explicitly set to False.

Return type:

dict

application.db.weather.set_user_excluded(username, exclude)#

Updates the exclusion status of a weather user and processes the updated user data.

Parameters:

  • username (str) – The username of the weather user.

  • exclude (bool) – The exclusion status to be set for the user.

Returns:

The processed user data after updating the exclusion status.

Return type:

dict

application.db.weather.update_user(user_data)#

Updates the weather information for a user in the database.

Parameters:

user_data (dict) – A dictionary containing user data with the following keys: - ‘username’ (str): The username of the user. - ‘lat’ (float): The latitude of the user’s location. - ‘lon’ (float): The longitude of the user’s location. - ‘max’ (dict): A dictionary with keys ‘disable’, ‘default’, and ‘value’ for the maximum temperature settings. - ‘min’ (dict): A dictionary with keys ‘disable’, ‘default’, and ‘value’ for the minimum temperature settings. - ‘_id’ (Any): The unique identifier of the user in the database.

Returns:

The user’s updated weather information.

Return type:

dict

Module contents#

This module provides direct access to the database and initializes various collections used by the application.

application.db.create_indexes()#

Create necessary indexes for the database collections to ensure efficient querying and proper functioning of the application.

Return type:

None

application.db.init_db(data_db_url='localhost', blob_path=None)#

Initialize the database connections and set up the necessary collections.

This function connects to the MongoDB instance specified by data_db_url and initializes various collections used by the application. It also sets the path for blob storage if provided.

Parameters:

  • data_db_url (str) – The URL of the MongoDB instance to connect to. Defaults to ‘localhost’.

  • blob_path (str, optional) – The path for blob storage. Defaults to None.

Return type:

None

Returns:

None

application.db.setup_db()#

Sets up the database by performing the following actions:

  1. Attempts to create a temporary admin user.

  2. Creates necessary indexes for the database to function properly.

Return type:

None

On this page

This Page