Skip to main content

ALDashboard.aldashboard

os​

shutil​

subprocess​

re​

ast​

requests​

calendar​

distributions​

UserModel​

Role​

UserRoles​

Github​

current_app​

flash​

redirect​

text​

bindparam​

func​

or_​

select​

joinedload​

Any​

List​

Tuple​

Dict​

Optional​

Callable​

Set​

math​

worker​

daconfig​

get_session_variables​

serializable_dict​

log​

DAFile​

DAObject​

DAList​

word​

DAFileList​

get_config​

user_has_privilege​

user_privileges​

DACloudStorage​

user_info​

user_logged_in​

get_user_info​

YAML​

StringIO​

werkzeug​

send_file​

secure_filename​

tempfile​

zipfile​

resources​

date​

datetime​

timedelta​

timezone​

_db_session_scope​

_get_db_session​

SavedFile​

__all__​

install_from_github_url​

def install_from_github_url(url: str,
                            branch: str = "",
                            pat: Optional[str] = None)

install_from_pypi​

def install_from_pypi(packagename: str)

github_url_to_package_name​

def github_url_to_package_name(url: str) -> str

Normalize a GitHub package URL to a python package name.

_get_github_version​

def _get_github_version(giturl: str) -> Optional[str]

Fetch the version from a GitHub repository's setup.py file.

Arguments​

Returns​

Version string if found, None otherwise.

_parse_version_tuple​

def _parse_version_tuple(version_str: str) -> Tuple

Convert version string to tuple of integers for comparison.

_compare_versions​

def _compare_versions(v1: str, v2: str) -> int

Compare two version strings.

Arguments​

  • v1 - First version
  • v2 - Second version

Returns​

-1 if v1 < v2, 0 if equal, 1 if v1 > v2

_get_latest_version​

def _get_latest_version(installed: str, pypi: Optional[str],
                        github: Optional[str]) -> Tuple[str, str]

Determine the latest available version among installed, pypi, and github versions.

Arguments​

  • installed - Installed version
  • pypi - PyPI version (if available)
  • github - GitHub version (if available)

Returns​

Tuple of (latest_version, source) where source is 'github', 'pypi', or 'installed'

_installed_distribution_versions​

def _installed_distribution_versions() -> Dict[str, str]

Return installed distribution versions keyed by casefolded package name.

_package_info_by_name​

def _package_info_by_name(
        pkg_versions: Optional[Dict[str, str]] = None,
        package_names: Optional[Set[str]] = None) -> Dict[str, Dict[str, Any]]

Return package metadata from docassemble package manager keyed by package name.

NOTE: We get the actual installed version from importlib.metadata (pkg_versions) rather than from the docassemble Package object, since the Package object's version attribute may be unreliable or represent a different value.

get_assemblyline_package_status​

def get_assemblyline_package_status(
        package_catalog: Dict[str, Dict[str, Any]]) -> List[Dict[str, Any]]

Return package status for AssemblyLine package installer choices.

Arguments​

  • package_catalog - mapping of github URL -> metadata (label/help/package_name/etc).

Returns​

List of dicts with fields including url, package_name, label, installed, version, can_update, latest_version, latest_source, and optional.

reset​

def reset(packagename="")

da_get_config_as_file​

def da_get_config_as_file()

da_get_config​

def da_get_config()

da_write_config​

def da_write_config(data: Dict)

get_user_count​

def get_user_count() -> int

Return the count of active (non-disabled) users on the server. Used to decide whether it's safe to load every user into a dropdown, or whether the list is too large to do that.

get_users_and_name_by_ids​

def get_users_and_name_by_ids(
        ids: List[int]) -> List[Tuple[int, str, str, str]]

Same shape as get_users_and_name(), but scoped to a specific list of user IDs instead of pulling every user in the database. Used to avoid loading the entire users table just to label a handful of already fetched session rows.

search_users_by_email​

def search_users_by_email(
        wordstart: str,
        limit: int = 20,
        exclude_privileged: bool = False) -> List[Tuple[int, str]]

Search for users whose email starts with the given text. Used by the input type: ajax field on large servers, so we never load the full user table - just return a handful of matches as the admin types.

speedy_get_users​

def speedy_get_users() -> List[Dict[int, str]]

Return a list of all users in the database. Possibly faster than get_user_list().

get_users_and_name​

def get_users_and_name(
    limit_to_non_admin_or_developers: bool = False
) -> List[Tuple[int, str, str, str]]

get_user_details​

def get_user_details(user_id: int) -> Optional[Dict]

Return a dictionary with details about a user, including privileges and MFA status.

Arguments​

  • user_id - The database ID of the user.

Returns​

A dictionary with user details, or None if the user is not found. Keys include: id, email, first_name, last_name, active, account_type, privileges (list of str), mfa_enabled (bool), mfa_type (str or None: "app", "sms", or None), last_login (datetime or None), session_count (int), and last_session_activity (datetime or None).

_current_user_permissions​

def _current_user_permissions() -> Set[str]

Return the current user's configured docassemble permissions.

can_access_user_sessions​

def can_access_user_sessions(user_id: Optional[int] = None) -> bool

Whether the current user may inspect another account's sessions.

Docassemble's access_sessions permission grants access to other users' sessions. The user-centered entry point also requires access_user_info. Non-admin support users may not use it to inspect protected accounts.

disable_user_mfa​

def disable_user_mfa(user_id: int) -> bool

Disable two-factor authentication for a user by clearing their OTP secret.

Arguments​

  • user_id - The database ID of the user.

Returns​

True if MFA was disabled, False if the user was not found or MFA was already disabled.

is_user_privileged​

def is_user_privileged(user_id: int) -> Optional[bool]

Check if a user has admin, developer, or cron privileges.

Arguments​

  • user_id - The database ID of the user.

Returns​

True if the user has any of the privileged roles, False if not, or None if the user could not be retrieved.

def get_password_reset_link(user_id: int) -> Optional[str]

Generate a password reset link for a specific user without sending an email.

Arguments​

  • user_id - The database ID of the user to generate the reset link for.

Returns​

A full password reset URL string, or None if the user is not found.

send_password_reset_email​

def send_password_reset_email(user_id: int) -> Dict[str, Any]

Ask docassemble's user manager to send its standard reset email.

This deliberately uses the same authorization and protected-account rules as :func:get_password_reset_link, and delegates token generation, message templates, and delivery to docassemble's built-in forgot-password flow.

recent_account_activity_report​

def recent_account_activity_report(
        days: int = 30,
        user_limit: int = 200,
        anonymous_limit: int = 100) -> Dict[str, Any]

Return recent login activity and grouped anonymous session activity.

Anonymous activity is grouped by docassemble's temporary user identifier. The interview/session tables do not contain a reliable historical source IP.

PLAYGROUND_SECTIONS​

_format_optional_datetime​

def _format_optional_datetime(value: Optional[datetime]) -> str

_epoch_to_datetime​

def _epoch_to_datetime(epoch_value: Optional[float]) -> Optional[datetime]

_latest_datetime​

def _latest_datetime(*values: Optional[datetime]) -> Optional[datetime]

_months_ago​

def _months_ago(months: int) -> datetime

_resolve_current_user_id​

def _resolve_current_user_id() -> Optional[int]

_user_session_summary​

def _user_session_summary(user_id: int) -> Dict[str, Any]

inactive_developer_login_summary​

def inactive_developer_login_summary(months: int = 18) -> Dict[str, Any]

Return a cheap login-only summary for the inactive developer report.

_playground_summary​

def _playground_summary(user_id: int) -> Dict[str, Any]

_developer_user_rows​

def _developer_user_rows() -> List[Any]

inactive_developer_account_report​

def inactive_developer_account_report(
        months: int = 18,
        requesting_user_id: Optional[int] = None) -> List[Dict[str, Any]]

Return developer accounts with no recent login, session, or playground activity.

delete_inactive_developer_accounts​

def delete_inactive_developer_accounts(
        user_ids: List[int],
        months: int = 18,
        delete_shared: bool = False,
        restart_if_needed: bool = True,
        requesting_user_id: Optional[int] = None) -> Dict[str, Any]

Delete selected stale developer accounts through docassemble's built-in cleanup.

SessionSearchCriteriaError Objects​

class SessionSearchCriteriaError(ValueError)

Raised when a session-search criterion is malformed or unsafe.

_SessionSearchPathPart Objects​

class _SessionSearchPathPart()

__init__​

def __init__(kind: str, value: Any)

_split_session_search_criterion​

def _split_session_search_criterion(line: str) -> Tuple[str, str]

Split path = value at an equals sign outside quotes and brackets.

_parse_session_search_path​

def _parse_session_search_path(path: str) -> List[_SessionSearchPathPart]

parse_session_search_criteria​

def parse_session_search_criteria(criteria_text: str) -> List[Dict[str, str]]

Parse one case-insensitive partial-match criterion per nonblank line.

build_session_search_criteria_text​

def build_session_search_criteria_text(
        variable_name: str,
        variable_value: str,
        *,
        use_advanced_filters: bool = False,
        advanced_criteria_text: str = "") -> str

Return the criteria text submitted to the session-search parser.

resolve_session_variable​

def resolve_session_variable(variables: Dict[str, Any], path: str) -> Any

Resolve a restricted Python-style path without using eval.

_display_session_value​

def _display_session_value(value: Any) -> str

_session_matches_criteria​

def _session_matches_criteria(filename: str, session_id: str,
                              criteria: List[Dict[str, str]]) -> bool

_iso_date_text​

def _iso_date_text(value: Any) -> str

format_session_users​

def format_session_users(session: Any, users_by_id: Dict[int, str]) -> str

Return a readable list of users associated with a session row or dict.

speedy_get_sessions​

def speedy_get_sessions(
        user_id: Optional[int] = None,
        temp_user_id: Optional[int] = None,
        filename: Optional[str] = None,
        filter_step1: bool = True,
        metadata_key_name: str = "metadata",
        start_date: Optional[Any] = None,
        end_date: Optional[Any] = None,
        search_criteria_text: Optional[str] = None) -> List[Tuple]

Return the 500 most recently active sessions, optionally tied to a registered user ID or an anonymous temporary-user ID.

Each session is a tuple with named columns: filename, user_id, modtime, key

get_session_details​

def get_session_details(session_id: str,
                        yaml_filename: Optional[str] = None) -> Dict[str, Any]

Get details for a session ID, optionally scoped to one interview filename.

get_upload_details​

def get_upload_details(file_id: Any) -> Optional[Dict[str, Any]]

Resolve a docassemble file number to its upload and owning session.

get_permitted_upload_details​

def get_permitted_upload_details(file_id: Any) -> Optional[Dict[str, Any]]

Resolve a file number and return it only when the current user may see it.

Both the upload's interview and its owning session are resolved from the database rather than trusted from the caller, so an action argument can't be used to get around the interview-viewer allow-list.

get_file_ids_associated_with_session​

def get_file_ids_associated_with_session(session_id: str,
                                         yaml_filename: Optional[str] = None
                                         ) -> List[int]

Get a list of file IDs associated with a given session ID.

get_files_associated_with_session​

def get_files_associated_with_session(
        session_id: str,
        yaml_filename: Optional[str] = None,
        privileged: Optional[bool] = False,
        uids: Optional[List[Any]] = None) -> List[Dict[str, Any]]

Return metadata for files associated with a session.

build_session_files_zip​

def build_session_files_zip(session_id: str,
                            yaml_filename: Optional[str] = None,
                            privileged: Optional[bool] = False,
                            uids: Optional[List[Any]] = None) -> Optional[str]

Build a temporary ZIP containing all files associated with a session. Returns the ZIP path or None if no files were found.

get_allowed_interview_filenames​

def get_allowed_interview_filenames() -> Optional[Set[str]]

Return the set of interview filenames the current user may look at sessions for, or None if the user is unrestricted (admins, developers, and users granted both access_user_info and access_sessions).

Non-privileged users are limited to the interviews listed under assembly line: interview viewers: in the configuration for one of the privileges they hold.

get_permitted_session_details​

def get_permitted_session_details(
        session_id: str,
        yaml_filename: Optional[str] = None) -> Optional[Dict[str, Any]]

Return get_session_details() for a session if the current user is allowed to view it, and None if the session doesn't exist or the user isn't allowed to see it.

The interview filename is always resolved from the database rather than trusted from the caller, so an action argument can't be used to get around the interview allow-list.

get_session_file_for_download​

def get_session_file_for_download(
        session_id: str,
        file_id: Any,
        yaml_filename: Optional[str] = None,
        privileged: Optional[bool] = False) -> Optional[DAFile]

Return a DAFile copy of one file belonging to the given session, suitable for response(file=...), or None if the file isn't part of that session.

The path comes from get_files_associated_with_session(), which has already resolved and verified it, so this serves exactly the files that were listed.

get_session_files_zip​

def get_session_files_zip(
        session_id: str,
        yaml_filename: Optional[str] = None,
        privileged: Optional[bool] = False) -> Optional[DAFile]

Return a DAFile ZIP of every file associated with a session, or None if the session has no files. The temporary ZIP written to disk is always removed.

download_file_by_id​

def download_file_by_id(file_number: str,
                        privileged: Optional[bool] = False,
                        uids: Optional[List[Any]] = None,
                        extension: Optional[str] = None,
                        filename_override: Optional[str] = None,
                        return_file_info: Optional[bool] = False)

Fetches a file by its ID and returns a Flask Response for downloading it.

Arguments​

  • file_number str - Raw file ID (may contain non-digit characters).
  • privileged bool - Whether to perform an admin/advocate lookup. Defaults to False.
  • uids list, optional - List of session UIDs for scoping the lookup. Defaults to None.
  • extension str, optional - File extension to force (e.g. "pdf"). Defaults to None.
  • filename_override str, optional - Specific filename to serve (e.g. "report.docx"). Defaults to None.
  • return_file_info bool, optional - If True, return metadata needed for a response(file=...) call instead of a Flask response.

Raises​

  • FileNotFoundError - If no record or filesystem path exists for the given ID, or the requested variant is missing on disk.

Returns​

Union[Response, Dict[str, Any]]: A Flask Response object from send_file, or file info dict.

dashboard_get_session_variables​

def dashboard_get_session_variables(session_id: str, filename: str)

Return the variables and steps for a given session ID and YAML filename in serializable dictionary format.

dashboard_find_session_filename​

def dashboard_find_session_filename(session_id: str) -> Optional[str]

Return the filename for a saved interview session key, when it can be identified unambiguously.

ALPackageInstaller Objects​

class ALPackageInstaller(DAObject)

Methods and state for installing AssemblyLine.

init​

def init(*pargs, **kwargs)

get_validated_github_username​

def get_validated_github_username(access_token: str)

Given a valid GitHub access_token, returns the username associated with it. Otherwise, adds one or more errors to the installer.

ErrorList Objects​

class ErrorList(DAList)

Contains ErrorLikeObjects so they can be recognized by docassemble.

init​

def init(*pargs, **kwargs)

ErrorLikeObject Objects​

class ErrorLikeObject(DAObject)

An object with a template_name that identifies the DALazyTemplate that will show its error. It can contain any other attributes so its template can access them as needed. DAObject doesn't seem to be enough to allow template definition.

init​

def init(*pargs, **kwargs)

install_fonts​

def install_fonts(the_font_files: DAFileList)

Install fonts to the server and restart both supervisor and unoconv.

list_installed_fonts​

def list_installed_fonts()

List the fonts installed on the server.

nicer_interview_filename​

def nicer_interview_filename(filename: str) -> str

Given a filename like docassemble.playground10ALWeaver:data/questions/assembly_line.yml, return a less cluttered name like: playground10ALWeaver:assembly_line

make_usage_rows​

def make_usage_rows(current_interview_usage: Optional[Dict[int, List[Dict]]],
                    nicer_fn: Callable[[str], str] = lambda x: x,
                    limit: int = 10) -> List[Dict]

Convert the nested current_interview_usage structure into a list of rows suitable for rendering in the template.

Arguments​

  • current_interview_usage - mapping of minute -> list of dicts with keys including 'filename', 'sessions', 'users'. Typically the output of dashboard_session_activity.
  • nicer_fn - callable that receives a filename and returns a display title.
  • limit - maximum number of rows to return (sorted by total recent sessions).

Returns​

A list of dicts with keys: filename, title, s_1, s_5, s_10, s_30, s_60, s_120, users, total

dashboard_session_activity​

def dashboard_session_activity(minutes_list=None,
                               limit: int = 10,
                               exclude_filenames=None)

Return a dict mapping each minutes value to a list of top interviews by session starts during the last N minutes. Each list contains dicts with keys: filename, sessions, users, title.

Arguments​

  • minutes_list - time windows to report on (default: [1, 5, 10, 30, 60, 120])

  • limit - max interviews per window (default: 10)

  • exclude_filenames - list of exact filenames or package prefixes to exclude. By default, excludes docassemble.ALDashboard: and entries from get_config("assembly line",{}).get("interview list",{}).get("exclude from interview list")

    Example return value:

  • {60 - [{'filename': 'docassemble.foo:data/questions/x.yml', 'sessions': 12, 'users': 9, 'title': 'foo:x'}, ...], ...}

compute_heatmap_styles​

def compute_heatmap_styles(rows, windows: Optional[tuple] = None)

Add inline styles for a log-scaled heatmap to each row dict.

list_question_files_in_package​

def list_question_files_in_package(package_name: str) -> Optional[List[str]]

List all the files in the 'data/questions' directory of a package.

Arguments​

  • package_name str - The name of the package to list files from.

Returns​

  • List[str] - A list of filenames in the 'data/questions' directory of the package.

list_question_files_in_docassemble_packages​

def list_question_files_in_docassemble_packages()

List all the files in the 'data/questions' directory of all docassemble packages.

Returns​

Dict[str, List[str]]: A dictionary where the keys are package names and the values are lists of filenames in the 'data/questions' directory of the package.

increment_index_value​

def increment_index_value(by: int = 5000,
                          index_name: str = "uploads_indexno_seq")

Increment the file index value in the database by a specified amount.

Arguments​

  • by int - The amount to increment the file index value by. Defaults to 5000.
  • index_name str - The name of the sequence to increment. Defaults to "uploads_indexno_seq".

get_current_index_value​

def get_current_index_value() -> int

Get the current value of the file index sequence.

Returns​

  • int - The current value of the file index sequence.

get_latest_s3_folder​

def get_latest_s3_folder(prefix: str = "files/") -> Optional[int]

Return the highest integer “folder” that exists directly under prefix, or None if there are no numeric folders at all.

• Uses the S3 LIST paginator, so it works for any number of prefixes. • Ignores non‑numeric folder names (e.g. files/tmp/, files/images/, …). • Requires only read permission for ListObjectsV2.

Example return value: 45237