core.sms_processor

Send SMS through ASPSMS

Adapted from `repoze.sendmail<https://github.com/repoze/repoze.sendmail>`_.

Usage::

qp = SmsQueueProcessor(sms_directory) qp.send_messages()

Attributes

log

MAX_SEND_TIME

Classes

SmsQueueProcessor

Functions

get_sms_queue_processor(→ SmsQueueProcessor | None)

Module Contents

core.sms_processor.log[source]
core.sms_processor.MAX_SEND_TIME = 10800[source]
class core.sms_processor.SmsQueueProcessor(path: str, username: str, password: str, originator: str | None = None)[source]
path[source]
username[source]
password[source]
originator = 'OneGov'[source]
url = 'https://json.aspsms.com/SendSimpleTextSMS'[source]
curl[source]
split(filename: str) tuple[str, str, str][source]

Returns the path, the name and the suffix of the given path.

message_files() tuple[str, Ellipsis][source]

Returns a tuple of full paths that need processing.

The file names in the directory usually look like this:

  • 0.1571822840.745629

  • 1.1571822743.595377

The part before the first dot is the batch number the rest is the timestamp at time of calling app.send_sms.

The messages are sorted by suffix, so by default the sorting happens from oldest to newest message.

send(numbers: Sequence[str], content: str) dict[str, Any] | None[source]

Sends the SMS and returns the API response on error.

On success this returns None.

send_request(parameters: onegov.core.types.JSON_ro) tuple[int, str][source]

Performes the API request using the given parameters.

parse(filename: str) tuple[tuple[str, Ellipsis] | None, str | None][source]
send_messages() None[source]
send_message(filename: str) None[source]
core.sms_processor.get_sms_queue_processor(app: onegov.core.framework.Framework, missing_path_ok: bool = False) SmsQueueProcessor | None[source]