ferrosoft.apps.ferrobase.services package

Submodules

ferrosoft.apps.ferrobase.services.captcha module

Lightweight image-based CAPTCHA used on public forms.

Generates a small arithmetic puzzle, renders it as a PNG, and stores the expected answer in the session for later verification.

class ferrosoft.apps.ferrobase.services.captcha.Captcha

Bases: object

Arithmetic-puzzle CAPTCHA bound to a Django session.

Each instance picks two random addends (or the fixed pair [1, 1] when the FERROBASE_INSECURE_CAPTCHA setting is enabled — an escape hatch for end-to-end tests). set_challenge() stores the expected answer in the session and verify() checks the submitted POST value.

FONT_SIZE = 16

INSECURE_OPTION = 'FERROBASE_INSECURE_CAPTCHA'

POST_KEY = 'probe'

SESSION_KEY = 'captcha_challenge'

make_image()

Create base64 encoded PNG image from text.

Return type:

str

set_challenge(request)

Store the expected answer (sum of the addends) in the session.

verify(request)

Return True when the POST submission matches the stored challenge.

Return type:

bool

ferrosoft.apps.ferrobase.services.cleanup module

class ferrosoft.apps.ferrobase.services.cleanup.CleanupService

Bases: object

Periodic database housekeeping orchestrator.

Runs a list of cleanup tasks against the master database and against each active tenant database. Tasks that raise are logged but do not abort the surrounding run, so unrelated cleanups continue to make progress.

DELETE_EXAMINATION_REQUEST_DAYS = 14

DELETE_TENANT_DAYS = 7

cleanup_master_database()

Run cleanup tasks in master database, log results.

cleanup_tenant_database(tenant)

Run cleanup tasks in tenant database, log results.

static deactivate_expired_tenants()

Mark tenants past their expiry_date as inactive; return the count.

Return type:

int

delete_empty_orgs()

Delete organisations that have no tenants left; return the count.

Return type:

int

static delete_expired_files()

Deletes expired StoredFile objects and their associated files for all tenants.

Return type:

int

delete_inactive_tenants()

Delete tenants that have been inactive for at least DELETE_TENANT_DAYS.

The two-phase pattern (first deactivate, then delete after a grace period) gives operators a window to reactivate a tenant before the record is gone.

Return type:

int

delete_old_examination_requests()

Delete examination requests older than DELETE_EXAMINATION_REQUEST_DAYS.

Returns 0 when the optional befundung app is not installed.

Return type:

int

ferrosoft.apps.ferrobase.services.conversion module

Scientific unit conversion and normalisation.

Provides direct unit-to-unit conversion via the MULTIPLIERS lookup table and category-based normalisation via NORMALIZED_UNITS. All arithmetic is done with decimal.Decimal and quantised to PRECISION to absorb precision loss introduced by Django ORM fields.

exception ferrosoft.apps.ferrobase.services.conversion.ConversionError

Bases: Exception

Raised when no multiplier exists for the requested source → target pair.

ferrosoft.apps.ferrobase.services.conversion.MULTIPLIERS: dict[tuple[UnitCursor, UnitCursor], Decimal] = {(UnitCursor(symbol='US cup', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('0.2365882365'), (UnitCursor(symbol='US fl oz', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('0.029573529563'), (UnitCursor(symbol='US gal', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('3.785411784'), (UnitCursor(symbol='US pt', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('0.473176473'), (UnitCursor(symbol='US qt', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('0.946352946'), (UnitCursor(symbol='acre ft', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('1233481.8375475'), (UnitCursor(symbol='bbl', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('119.240471196'), (UnitCursor(symbol='bu', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('35.23907016688'), (UnitCursor(symbol='cb', system=UnitSystem.IMPERIAL), UnitCursor(symbol='km', system=UnitSystem.METRIC)): Decimal('0.1852'), (UnitCursor(symbol='cb', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='km', system=UnitSystem.METRIC)): Decimal('0.219456'), (UnitCursor(symbol='ch', system=UnitSystem.IMPERIAL), UnitCursor(symbol='km', system=UnitSystem.METRIC)): Decimal('0.0201168'), (UnitCursor(symbol='cu ft', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('28.316846592'), (UnitCursor(symbol='cu in', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('0.016387064'), (UnitCursor(symbol='cu yd', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('764.554857984'), (UnitCursor(symbol='cwt', system=UnitSystem.IMPERIAL), UnitCursor(symbol='t', system=UnitSystem.METRIC)): Decimal('0.05080234544'), (UnitCursor(symbol='cwt', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='t', system=UnitSystem.METRIC)): Decimal('0.045359237'), (UnitCursor(symbol='dry bbl', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('115.627123584'), (UnitCursor(symbol='dw t', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='t', system=UnitSystem.METRIC)): Decimal('0.00000155517384'), (UnitCursor(symbol='fl oz', system=UnitSystem.IMPERIAL), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('0.028413062500'), (UnitCursor(symbol='ft', system=UnitSystem.IMPERIAL), UnitCursor(symbol='km', system=UnitSystem.METRIC)): Decimal('0.0003048'), (UnitCursor(symbol='ft', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='km', system=UnitSystem.METRIC)): Decimal('0.0003048'), (UnitCursor(symbol='ftm', system=UnitSystem.IMPERIAL), UnitCursor(symbol='km', system=UnitSystem.METRIC)): Decimal('0.0018288'), (UnitCursor(symbol='ftm', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='km', system=UnitSystem.METRIC)): Decimal('0.0018288'), (UnitCursor(symbol='fur', system=UnitSystem.IMPERIAL), UnitCursor(symbol='km', system=UnitSystem.METRIC)): Decimal('0.201168'), (UnitCursor(symbol='g', system=UnitSystem.METRIC), UnitCursor(symbol='kg', system=UnitSystem.METRIC)): Decimal('0.001'), (UnitCursor(symbol='g', system=UnitSystem.METRIC), UnitCursor(symbol='t', system=UnitSystem.METRIC)): Decimal('0.000001'), (UnitCursor(symbol='gal', system=UnitSystem.IMPERIAL), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('4.54609'), (UnitCursor(symbol='gal', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('4.40488377086'), (UnitCursor(symbol='gi', system=UnitSystem.IMPERIAL), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('0.1420653125'), (UnitCursor(symbol='hhd', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('238.480942392'), (UnitCursor(symbol='kWh', system=UnitSystem.METRIC), UnitCursor(symbol='MJ', system=UnitSystem.METRIC)): Decimal('3.6'), (UnitCursor(symbol='kg', system=UnitSystem.METRIC), UnitCursor(symbol='g', system=UnitSystem.METRIC)): Decimal('1000'), (UnitCursor(symbol='kg', system=UnitSystem.METRIC), UnitCursor(symbol='t', system=UnitSystem.METRIC)): Decimal('0.001'), (UnitCursor(symbol='kgCO2e', system=UnitSystem.METRIC), UnitCursor(symbol='gCO2e', system=UnitSystem.METRIC)): Decimal('1000'), (UnitCursor(symbol='km', system=UnitSystem.METRIC), UnitCursor(symbol='m', system=UnitSystem.METRIC)): Decimal('1000'), (UnitCursor(symbol='lb t', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='t', system=UnitSystem.METRIC)): Decimal('0.0003732417216'), (UnitCursor(symbol='lb', system=UnitSystem.IMPERIAL), UnitCursor(symbol='t', system=UnitSystem.METRIC)): Decimal('0.00045359237'), (UnitCursor(symbol='lb', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='t', system=UnitSystem.METRIC)): Decimal('0.00045359237'), (UnitCursor(symbol='le', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='km', system=UnitSystem.METRIC)): Decimal('4.828032'), (UnitCursor(symbol='lea', system=UnitSystem.IMPERIAL), UnitCursor(symbol='km', system=UnitSystem.METRIC)): Decimal('4.828032'), (UnitCursor(symbol='long cwt', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='t', system=UnitSystem.METRIC)): Decimal('0.05080234544'), (UnitCursor(symbol='m', system=UnitSystem.METRIC), UnitCursor(symbol='km', system=UnitSystem.METRIC)): Decimal('0.001'), (UnitCursor(symbol='mi', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='km', system=UnitSystem.METRIC)): Decimal('1.609344'), (UnitCursor(symbol='mile', system=UnitSystem.IMPERIAL), UnitCursor(symbol='km', system=UnitSystem.METRIC)): Decimal('1.609344'), (UnitCursor(symbol='nmi', system=UnitSystem.IMPERIAL), UnitCursor(symbol='km', system=UnitSystem.METRIC)): Decimal('1.852'), (UnitCursor(symbol='nmi', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='km', system=UnitSystem.METRIC)): Decimal('1.852'), (UnitCursor(symbol='oil bbl', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('158.987294928'), (UnitCursor(symbol='oz t', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='t', system=UnitSystem.METRIC)): Decimal('0.0000311034768'), (UnitCursor(symbol='oz', system=UnitSystem.IMPERIAL), UnitCursor(symbol='t', system=UnitSystem.METRIC)): Decimal('2.8349523125E-8'), (UnitCursor(symbol='oz', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='t', system=UnitSystem.METRIC)): Decimal('0.000028349523125'), (UnitCursor(symbol='pk', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('8.80976754172'), (UnitCursor(symbol='pot', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('1.892705892'), (UnitCursor(symbol='pt', system=UnitSystem.IMPERIAL), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('0.56826125'), (UnitCursor(symbol='pt', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('0.5506104713575'), (UnitCursor(symbol='qr', system=UnitSystem.IMPERIAL), UnitCursor(symbol='t', system=UnitSystem.METRIC)): Decimal('0.01270058636'), (UnitCursor(symbol='qt', system=UnitSystem.IMPERIAL), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('1.1365225'), (UnitCursor(symbol='qt', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='L', system=UnitSystem.METRIC)): Decimal('1.101220942715'), (UnitCursor(symbol='st', system=UnitSystem.IMPERIAL), UnitCursor(symbol='t', system=UnitSystem.METRIC)): Decimal('0.00635029318'), (UnitCursor(symbol='st', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='t', system=UnitSystem.METRIC)): Decimal('0.90718474'), (UnitCursor(symbol='t', system=UnitSystem.METRIC), UnitCursor(symbol='g', system=UnitSystem.METRIC)): Decimal('1000000'), (UnitCursor(symbol='t', system=UnitSystem.METRIC), UnitCursor(symbol='kg', system=UnitSystem.METRIC)): Decimal('1000'), (UnitCursor(symbol='ton', system=UnitSystem.IMPERIAL), UnitCursor(symbol='t', system=UnitSystem.METRIC)): Decimal('1.0160469088'), (UnitCursor(symbol='ton', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='t', system=UnitSystem.METRIC)): Decimal('1.0160469088'), (UnitCursor(symbol='yd', system=UnitSystem.IMPERIAL), UnitCursor(symbol='km', system=UnitSystem.METRIC)): Decimal('0.0009144'), (UnitCursor(symbol='yd', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='km', system=UnitSystem.METRIC)): Decimal('0.0009144')}

Define multipliers for conversion from one unit to another unit.

ferrosoft.apps.ferrobase.services.conversion.NON_METRIC_VOLUME_UNITS = [UnitCursor(symbol='cu in', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='cu ft', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='cu yd', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='acre ft', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='US fl oz', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='US cup', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='US pt', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='US qt', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='pot', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='US gal', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='bbl', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='oil bbl', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='hhd', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='pt', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='qt', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='gal', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='pk', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='bu', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='dry bbl', system=UnitSystem.US_CUSTOMARY), UnitCursor(symbol='fl oz', system=UnitSystem.IMPERIAL), UnitCursor(symbol='gi', system=UnitSystem.IMPERIAL), UnitCursor(symbol='pt', system=UnitSystem.IMPERIAL), UnitCursor(symbol='qt', system=UnitSystem.IMPERIAL), UnitCursor(symbol='gal', system=UnitSystem.IMPERIAL)]

List of non-metric units which can be converted to liter.

ferrosoft.apps.ferrobase.services.conversion.NORMALIZED_UNITS: dict[tuple[UnitCategory, UnitCursor | None], UnitCursor] = {(UnitCategory.DISTANCE, None): UnitCursor(symbol='km', system=UnitSystem.METRIC), (UnitCategory.EMISSION_FACTOR, UnitCursor(symbol='gCO2e/MJ', system=UnitSystem.METRIC)): UnitCursor(symbol='gCO2e/MJ', system=UnitSystem.METRIC), (UnitCategory.EMISSION_FACTOR, UnitCursor(symbol='kgCO2e/kg', system=UnitSystem.METRIC)): UnitCursor(symbol='kgCO2e/kg', system=UnitSystem.METRIC), (UnitCategory.EMISSION_VALUE, None): UnitCursor(symbol='gCO2e', system=UnitSystem.METRIC), (UnitCategory.MASS, None): UnitCursor(symbol='t', system=UnitSystem.METRIC), (UnitCategory.NATIVE_CONSUMPTION, None): UnitCursor(symbol='kg', system=UnitSystem.METRIC), (UnitCategory.NATIVE_CONSUMPTION, UnitCursor(symbol='kWh', system=UnitSystem.METRIC)): UnitCursor(symbol='MJ', system=UnitSystem.METRIC)}

Define which unit to convert to for a given unit category.

The second tuple member specifies the source unit. If it is None, the target unit serves as the default for any source unit. If it’s not None, the target unit takes precedence over the default target unit.

Advanced special casing is done in normalize_unit.

exception ferrosoft.apps.ferrobase.services.conversion.NormalizationError

Bases: ConversionError

Raised when a value cannot be normalised (missing unit or density).

ferrosoft.apps.ferrobase.services.conversion.PRECISION = Decimal('1.0000000000000000')

Working precision (16 decimal places) used to quantise intermediate values.

ferrosoft.apps.ferrobase.services.conversion.convert_unit(value, target_unit)

Convert value to target_unit using MULTIPLIERS.

Parameters:

• value (DecimalWithUnit) – Source quantity carrying its own unit.

• target_unit (UnitCursor) – Desired target unit.

Returns:

value expressed in target_unit. When the

source and target units already match, value is returned as a shallow copy without arithmetic.

Return type:

DecimalWithUnit

Raises:

ConversionError – If no direct multiplier exists for the (source_unit, target_unit) pair.

ferrosoft.apps.ferrobase.services.conversion.normalize_unit(category, value, *, density=None)

Convert value to a blessed unit of the given unit category.

See NORMALIZED_UNITS for blessed units. In addition, this function may handle some special casing pertaining to the unit category.

Parameters:

• category (UnitCategory) – Unit category of the blessed unit to convert to.

• value (DecimalWithUnit) – Value to convert.

• density (Decimal | None) – Density must be specified for NATIVE_CONSUMPTION category, if the source unit is a volume unit (e.g. liter or fluid ounce).

Return type:

DecimalWithUnit

ferrosoft.apps.ferrobase.services.help module

ferrosoft.apps.ferrobase.services.help.get_help_url(topic)

Build a context-help URL by appending #topic to the configured base.

Parameters:

topic (str) – Help anchor name to navigate to within the help site.

Returns:

Absolute URL pointing at the topic.

Return type:

str

Raises:

Exception – Propagates from require_settings() when the FERROBASE_HELP_BASE_URL setting is missing.

ferrosoft.apps.ferrobase.services.idgenerator module

class ferrosoft.apps.ferrobase.services.idgenerator.IdGenerator

Bases: ABC

Abstract sequence-based identifier generator.

Concrete subclasses bind a sequence to a backing store and produce formatted IDs combining the configured prefix and zero-padded sequence number.

abstractmethod classmethod create(name)

Create or load the generator for the sequence identified by name.

Return type:

IdGenerator

classmethod get_instance(name)

Return the default IdGeneratorPostgres for name.

abstractmethod next_id()

Return the next formatted identifier from the underlying sequence.

Return type:

str

class ferrosoft.apps.ferrobase.services.idgenerator.IdGeneratorPostgres(config)

Bases: IdGenerator

PostgreSQL-backed ID generator using CREATE SEQUENCE.

classmethod create(name)

Persist a configuration if missing and ensure the SQL sequence exists.

The IdSequenceConfig row is created in the master database with a default prefix equal to name.label and seven digits of padding. The corresponding PostgreSQL sequence is created in the tenant database via CREATE SEQUENCE IF NOT EXISTS.

Return type:

IdGeneratorPostgres

next_id()

Allocate the next sequence value and format it as prefix + padded_number.

Return type:

str

ferrosoft.apps.ferrobase.services.onboarding module

class ferrosoft.apps.ferrobase.services.onboarding.SignupRequest(email, company, first_name, last_name, language)

Bases: object

Inbound signup payload captured from the public registration form.

company: str

email: str

first_name: str

language: str

last_name: str

ferrosoft.apps.ferrobase.services.onboarding.activate_account(activation, user_password)

Complete onboarding by creating the first user and activating the tenant.

The user is granted the broad EMIFLOW_USER/EMIFLOW_MANAGER/ USER_MANAGER roles so the new organisation has at least one full administrator. Everything runs inside a single atomic transaction; if any step raises the tenant remains inactive and the activation request is preserved for retry.

Parameters:

• activation (AccountActivationRequest) – The activation record discovered via the email link.

• user_password (str) – Cleartext password chosen by the user — hashed by Django’s user-creation machinery.

async ferrosoft.apps.ferrobase.services.onboarding.initiate_signup(request: SignupRequest, frontend_uri: str) → bool

Create tenant + organisation and send the account-activation email.

The tenant is created inactive with a two-day expiry; if the user fails to follow the activation link in time the cleanup service tears it down. The first user is created later by activate_account().

Parameters:

• request (SignupRequest) – Sanitised signup payload.

• frontend_uri (str) – Absolute base URL of the frontend, used to build the activation link embedded in the welcome email.

Returns:

True on success, False when the request is silently

refused (existing user, mail-server unreachable, race-condition IntegrityError).

Return type:

bool

ferrosoft.apps.ferrobase.services.sse module

Server Side Events (SSE) client for Python.

Provides a generator of SSE received through an existing HTTP response.

class ferrosoft.apps.ferrobase.services.sse.SSEClient(event_source, char_enc='utf-8')

Bases: object

Implementation of a SSE client.

See http://www.w3.org/TR/2009/WD-eventsource-20091029/ for the specification.

close()

Manually close the event source stream.

events()

ferrosoft.apps.ferrobase.services.static_files module

ferrosoft.apps.ferrobase.services.static_files.static_file_name(app_name, file_name)

Find static file inside emiflow app.

Raises:

RuntimeError if the file is not found. –

Return type:

str

ferrosoft.apps.ferrobase.services.taskqueue module

class ferrosoft.apps.ferrobase.services.taskqueue.CheapyTaskQueue(client, base_url='http://localhost:8090', request_timeout=5)

Bases: TaskQueue

TaskQueue implementation for the CheapyQ service.

DEFAULT_BASE_URL = ''

async enqueue(func, *args, task_stream=None, result_group=None, **kwargs)

POST a base64/pickle-encoded task to /task and return its public ID.

Return type:

int

async stream_results(result_group, count, discard_result=False)

Subscribe to CheapyQ’s SSE /status endpoint for result_group.

Each successfully parsed event becomes a TaskResult. Events whose payload fails to parse are logged plus dumped to stderr for debugging and skipped. The generator exits after count results have been yielded.

Parameters:

• result_group (str) – Identifier shared with the corresponding enqueue calls.

• count (int) – Total number of results to await before exiting.

• discard_result (bool) – When True, the result field is left as None and pickle deserialisation is skipped (useful when only the success/error flag is interesting).

Return type:

AsyncIterator[TaskResult]

class ferrosoft.apps.ferrobase.services.taskqueue.TaskQueue

Bases: ABC

TaskQueue provides the low-level interface for interacting with the task queue.

async enqueue(func, *args, task_stream=None, result_group=None, **kwargs)

Submit func to the queue and return its numeric public ID.

Return type:

int

async stream_results(result_group, count, discard_result=False)

Yield TaskResult records for result_group until count arrive.

Return type:

AsyncIterator[TaskResult]

exception ferrosoft.apps.ferrobase.services.taskqueue.TaskQueueError

Bases: RuntimeError

Raised when the task queue cannot enqueue, stream or deserialise a task.

class ferrosoft.apps.ferrobase.services.taskqueue.TaskResult(id, status, creation_time, finish_time=None, error=None, result=None)

Bases: object

Per-task status payload streamed from the queue.

Carries the public ID, current TaskStatus, creation/finish timestamps and either a deserialised result or an error string.

creation_time: datetime

error: str | None = None

finish_time: datetime | None = None

id: UUID

result: Any | None = None

status: TaskStatus

class ferrosoft.apps.ferrobase.services.taskqueue.TaskStatus(*values)

Bases: StrEnum

Lifecycle status reported back by the task queue for each task.

ACTIVE = 'active'

ERROR = 'error'

NEW = 'new'

QUEUED = 'queued'

SUCCESS = 'success'

UNKNOWN = ''

async ferrosoft.apps.ferrobase.services.taskqueue.a_background_task(func, *args, add_language=True, result_group=None, task_stream=None, task_queue=Provide['task_queue'], **kwargs)

Schedules a task to be executed asynchronously.

Parameters:

• func (str) – Fully qualified Python function name.

• *args – Arguments passed to func.

• **kwargs – Keyword arguments passed to func.

• task_queue (TaskQueue) – The task queue implementation to use.

• add_language (bool) – If true, current language is added to function kwargs._language, which can be used in conjunction with locale_aware_task to enable user locale for as task.

• result_group (str | None) – Create tasks in the named group, results available via wait_for_results.

• task_stream (str | None) – Sends tasks to the named stream, to partition task execution. Avoids congestion. If the stream is not configured in the task queue, the default stream is used.

Return type:

int

Returns: Task ID (for diagnostic purposes)

async ferrosoft.apps.ferrobase.services.taskqueue.a_background_task_tenant(func, *args, task_queue=Provide['task_queue'], **kwargs)

Schedules a task to be executed asynchronously.

This function is like background_task, except that it also passes the keyword argument tenant to func, with the current tenant ID. The function should use the decorator tenant_task to opt in to automatic tenant database activation.

async ferrosoft.apps.ferrobase.services.taskqueue.a_wait_for_finish(*, result_group=None, count=None, task_queue=Provide['task_queue'])

Block until count tasks of result_group finish; discard payloads.

async ferrosoft.apps.ferrobase.services.taskqueue.a_wait_for_results(*, result_group, count=None, task_queue=Provide['task_queue'])

Stream count task results from result_group and return their values.

Returns:

The deserialised result of each task in the order

CheapyQ delivered them.

Return type:

List[Any]

async ferrosoft.apps.ferrobase.services.taskqueue.a_wait_for_success(*, result_group=None, count=None, task_queue=Provide['task_queue'])

Return True only when every awaited task returned a truthy boolean result.

Return type:

bool

ferrosoft.apps.ferrobase.services.taskqueue.background_task(*args, **kwargs)

Synchronous facade for async background_task.

Return type:

int

ferrosoft.apps.ferrobase.services.taskqueue.background_task_tenant(*args, **kwargs)

Synchronous facade for async background_task_tenant.

async ferrosoft.apps.ferrobase.services.taskqueue.task_runner(*args, **kwargs)

Recommended task runner function invoked by Cheapyq.

Task runners are an implementation detail of Cheapyq.

Its main use is automatic activation of language based on _language keyword argument, which is passed implicitly by background_task.

ferrosoft.apps.ferrobase.services.taskqueue.tenant_task()

Decorator activating the tenant context around a worker task.

The wrapper pops the tenant kwarg (added by a_background_task_tenant()), looks up the tenant and enters active_tenant() for the duration of the call. Missing or malformed tenant IDs raise TaskQueueError.

ferrosoft.apps.ferrobase.services.taskqueue.wait_for_finish(**kwargs)

Synchronous facade for async wait_for_results.

ferrosoft.apps.ferrobase.services.taskqueue.wait_for_results(**kwargs)

Synchronous facade for async wait_for_results.

Return type:

List[Any]

ferrosoft.apps.ferrobase.services.taskqueue.wait_for_success(**kwargs)

Synchronous facade for async wait_for_results.

Return type:

bool

ferrosoft.apps.ferrobase.services.tenant_mgm module

Tenant provisioning: PostgreSQL role/database creation, migrations and fixtures.

class ferrosoft.apps.ferrobase.services.tenant_mgm.TenantOptions(id, name, active, expiry_date, language, organization, fixtures, db_name=None, db_host=None, db_username=None, db_port=None, db_password=None)

Bases: object

active: bool

db_host: str | None = None

Override default database host (not tested).

db_name: str | None = None

Override default database name (not tested).

db_password: str | None = None

Override default database password (not tested).

db_port: str | None = None

Override default database port (not tested).

db_username: str | None = None

Override default database username (not tested).

expiry_date: datetime | None

fixtures: list[str]

Fixtures to import. The string LANGUAGE_CODE is replaced with language code

id: str | None

ID of the tenant or None to autogenerate

language: str

Two letter language code

name: str

organization: Organization

Organization to add tenant to

ferrosoft.apps.ferrobase.services.tenant_mgm.add_tenant(options)

Provision a new tenant database and register it in the master DB.

Must be called with the default (master) database connection active. The function:

1. Inserts the Tenant row as inactive so subsequent failures do not yield a half-provisioned, user-visible tenant.

2. Creates a PostgreSQL role and a tenant-specific database owned by it.

3. Runs migrate and the configured fixtures inside the new database.

4. Runs the registered tenant updaters and built-in fixture imports.

5. Flips the tenant’s active flag to options.active once everything succeeded.

Parameters:

options (TenantOptions) – Provisioning parameters; an empty id triggers UUIDv7 generation, and an empty db_password causes a random 36-char password to be generated.

Returns:

The newly created tenant record.

Return type:

Tenant

ferrosoft.apps.ferrobase.services.update module

class ferrosoft.apps.ferrobase.services.update.LoggingStatusWriter(logger)

Bases: StatusWriter

Status writer forwarding messages to a logging.Logger.

error(message)

Log message at ERROR level.

notice(message)

Log message at INFO level.

success(message)

Log message at INFO level.

class ferrosoft.apps.ferrobase.services.update.StatusWriter

Bases: ABC

Writer used by updater functions to send success or error messages.

abstractmethod error(message)

Report an error encountered by an updater.

abstractmethod notice(message)

Report informational progress from an updater.

abstractmethod success(message)

Report a successful updater step.

class ferrosoft.apps.ferrobase.services.update.Updater

Bases: object

Update database resources with data from current release.

By default, only Ferrobase resources are updated. Apps need to call add_global_updater/add_tenant_updater to register their updaters.

classmethod add_global_updater(*updaters)

Register updater callables to run against the master database.

classmethod add_tenant_updater(*updaters)

Register updater callables to run inside each tenant database.

classmethod init()

Seed the default updater registries on first call.

Idempotent: re-invocations do nothing if updaters are already registered, so apps that add their own updaters before import are not overwritten.

run(writer)

Run all registered updaters: globals first, then per-tenant updaters.

run_global_updaters(writer)

Invoke each registered global updater with tenant=None.

run_tenant_updaters(writer, tenant)

Invoke each registered tenant updater bound to tenant.

class ferrosoft.apps.ferrobase.services.update.WriterNoop

Bases: StatusWriter

Discard all status messages (useful when running updaters silently).

error(message)

Drop the message.

notice(message)

Drop the message.

success(message)

Drop the message.

ferrosoft.apps.ferrobase.services.update.create_permission_sets(writer, tenant)

Sync Django groups + permissions from the role registry.

Only runs in the master database (tenant is None); permission sets are global and not duplicated per tenant.

ferrosoft.apps.ferrobase.services.update.create_registered_models(writer, tenant)

Persist the in-memory ModelRegistry to the target database.

ferrosoft.apps.ferrobase.services.user_mgm module

class ferrosoft.apps.ferrobase.services.user_mgm.UserOptions(username, email, first_name, last_name, organization, password=None, roles=<factory>)

Bases: object

Parameters for add_user().

password may be None for SSO-only users; roles defaults to an empty list when no role assignment is required.

email: str

first_name: str

last_name: str

organization: Organization

password: str | None = None

roles: List[Role]

username: str

ferrosoft.apps.ferrobase.services.user_mgm.add_user(options)

Create a User and optionally assign the configured roles.

Wraps User.objects.create_user() (so passwords are hashed by Django) and then delegates to user.assign_roles when any roles are listed.

Return type:

User

ferrosoft.apps.ferrobase.services.zitadel module

HTTP REST client and request/response dataclasses for the ZITADEL IAM API v2.

class ferrosoft.apps.ferrobase.services.zitadel.Admin(userId=None, human=None, roles=<factory>)

Bases: object

Organisation administrator entry (existing userId or inline human).

human: Human | None = None

roles: List[str]

userId: str | None = None

class ferrosoft.apps.ferrobase.services.zitadel.CodeAction(urlTemplate=None)

Bases: object

Represents actions like sendCode which may carry an optional urlTemplate.

urlTemplate: str | None = None

class ferrosoft.apps.ferrobase.services.zitadel.Email(email, sendCode=None, returnCode=None, isVerified=None)

Bases: object

Email object with optional verification flow descriptors.

email: str

isVerified: bool | None = None

returnCode: EmptyObject | None = None

sendCode: CodeAction | None = None

class ferrosoft.apps.ferrobase.services.zitadel.EmptyObject

Bases: object

Represents an empty JSON object {}.

to_dict()

Serialise as an empty dict.

Return type:

Dict[str, Any]

class ferrosoft.apps.ferrobase.services.zitadel.Human(email, profile, userId=None, username=None, phone=None, metadata=<factory>, password=None, idpLinks=<factory>)

Bases: object

Human-user payload sent when creating Zitadel users.

email: Email

idpLinks: List[IDPLink]

metadata: List[MetadataItem]

password: Password | None = None

phone: Phone | None = None

profile: Profile

userId: str | None = None

username: str | None = None

class ferrosoft.apps.ferrobase.services.zitadel.IDPLink(idpId=None, userId=None, userName=None)

Bases: object

Identity-provider linkage referencing an external user.

idpId: str | None = None

userId: str | None = None

userName: str | None = None

class ferrosoft.apps.ferrobase.services.zitadel.MetadataItem(key, value)

Bases: object

Single user-metadata entry (value must be base64-encoded).

key: str

value: str

class ferrosoft.apps.ferrobase.services.zitadel.Organization(name, admins=<factory>, orgId=None)

Bases: object

Organisation payload combining name, admins and optional orgId.

admins: List[Admin]

name: str

orgId: str | None = None

to_dict()

Serialise to a JSON-ready dict via _to_dict().

Return type:

Dict[str, Any]

class ferrosoft.apps.ferrobase.services.zitadel.Password(password, changeRequired=None)

Bases: object

Initial password assignment, optionally forcing change on next login.

changeRequired: bool | None = None

password: str

class ferrosoft.apps.ferrobase.services.zitadel.Phone(phone, sendCode=None, returnCode=None, isVerified=None)

Bases: object

Phone object with optional verification flow descriptors.

isVerified: bool | None = None

phone: str

returnCode: EmptyObject | None = None

sendCode: CodeAction | None = None

class ferrosoft.apps.ferrobase.services.zitadel.Profile(givenName, familyName, nickName=None, displayName=None, preferredLanguage=None, gender=None)

Bases: object

User profile attributes as expected by the Zitadel Human payload.

displayName: str | None = None

familyName: str

gender: str | None = None

givenName: str

nickName: str | None = None

preferredLanguage: str | None = None

class ferrosoft.apps.ferrobase.services.zitadel.ZitadelClient(base_url, api_token, timeout=10.0, session=None)

Bases: object

Minimal HTTP REST client for ZITADEL API v2 focusing on “Create an Organization”.

Authentication:

• OAuth2 Client Credentials via /oauth/v2/token

• Scopes are passed as a space-separated string; for service users you typically need the audience scope “urn:zitadel:iam:org:project:id:<PROJECT_ID>:aud” among others, depending on your setup.

Endpoints:

• Token: {base_url}/oauth/v2/token

• Organizations: {base_url}/v2/organizations (Create via POST)

create_authorization(user_id, organization_id, project_id, project_grant_id, *role_keys)

Assign role_keys of a project grant to a specific user.

create_organization(organization)

Create an organization.

Parameters:

organization (Organization) – The organization to create.

Return type:

Dict[str, Any]

Returns:

The API response JSON as a dict.

Raises:

ZitadelClientError on HTTP errors or unexpected responses. –

create_project_grant(project_id, organization_id, *roles)

Grant roles of project_id to organization_id.

classmethod from_settings(**kwargs)

exception ferrosoft.apps.ferrobase.services.zitadel.ZitadelClientError

Bases: Exception

Raised for non-successful responses or OAuth failures.

ferrosoft.apps.ferrobase.services.zitadel.iam_url(path)

Return path joined onto the configured OIDC_OP_BASE_URL.

Return type:

str

ferrosoft.apps.ferrobase.services.zitadel.zitadel_logout(request)

Build the OIDC end-session URL that signs the user out of Zitadel.