core.framework

The Framework provides a base Morepath application that offers certain features for applications deriving from it:

• Virtual hosting in conjunction with onegov.server.

• Access to an SQLAlchemy session bound to a specific Postgres schema.

• A cache backed by redis, shared by multiple processes.

• An identity policy with basic rules, permissions and role.

• The ability to serve static files and css/js assets.

Using the framework does not really differ from using Morepath:

from onegov.core.framework import Framework

class MyApplication(Framework):
    pass

Attributes

_P
_T

Classes

Framework

Baseclass for Morepath OneGov applications.

Functions

get_webasset_url(→ str)	The webassets url needs to be unique so we can fix it before
get_js_filter(→ str)
get_css_filter(→ str)
get_jsx_filter(→ str)
fix_webassets_url_factory(→ Callable[[CoreRequest], ...)
get_retry_attempts(→ int)
get_cronjobs_enabled(→ bool)	If this value is set to False, all cronjobs are disabled. Only use
default_content_security_policy(...)	The default content security policy used throughout OneGov.
default_policy_apply_factory(...)	Adds the content security policy report settings from the yaml.
http_conflict_tween_factory(→ Callable[[CoreRequest], ...)
activate_session_manager_factory(...)	Activate the session manager before each transaction.
close_session_after_request_factory(...)	Closes the session after each request.
current_language_tween_factory(...)
spawn_cronjob_thread_tween_factory(...)

Module Contents

core.framework._P

core.framework._T

class core.framework.Framework

Bases: more.transaction.TransactionApp, more.webassets.WebassetsApp, onegov.core.orm.cache.OrmCacheApp, more.content_security.ContentSecurityApp, onegov.server.Application

Baseclass for Morepath OneGov applications.

request_class: type[morepath.request.Request]

The class of the Request to create. Must be a subclass of morepath.Request.

By default the request class is morepath.Request

dsn: str | None = None

schema: str = None

form

cronjob

static_directory

template_variables

same_site_cookie_policy: str | None = 'Lax'

request_cache: dict[str, Any]

schema_cache: dict[str, Any]

_all_schema_caches: dict[str, Any]

property version: str

__call__(environ: _typeshed.wsgi.WSGIEnvironment, start_response: _typeshed.wsgi.StartResponse) → collections.abc.Iterable[bytes]

This app as a WSGI application.

See the WSGI_ spec for more information.

Uses App.request() to generate a morepath.Request instance, then uses meth:App.publish get the morepath.Response instance.

Parameters:

• environ – WSGI environment

• start_response – WSGI start_response

Returns:

WSGI iterable.

with_query_report(fn: Callable[_P, _T]) → Callable[_P, _T]

with_profiler(fn: Callable[_P, _T]) → Callable[_P, _T]

with_request_cache(fn: Callable[_P, _T]) → Callable[_P, _T]

with_print_exceptions(fn: Callable[_P, _T]) → Callable[_P, _T]

clear_request_cache() → None

property modules: onegov.core.utils.Bunch

Provides access to modules used by the Framework class. Those modules cannot be included at the top because they themselves usually include the Framework.

Admittelty a bit of a code smell.

property metadata: core.metadata.Metadata

property has_database_connection: bool

onegov.core has good integration for Postgres using SQLAlchemy, but it doesn’t require a connection.

It’s possible to have Onegov applications using a different database or not using one at all.

property has_filestorage: bool

Returns true if fs is available.

handle_exception(exception: BaseException, environ: _typeshed.wsgi.WSGIEnvironment, start_response: _typeshed.wsgi.StartResponse) → Iterable[bytes]

Stops database connection errors from bubbling all the way up to our exception handling services (sentry.io).

configure_application(**cfg: Any) → None

Configures the application. This function calls all methods on the current class which start with configure_, passing the configuration as keyword arguments.

The core itself supports the following parameters. Additional parameters are made available by extra configure_ methods.

Dsn:

The database connection to use. May be None.

See onegov.core.orm.session_manager.setup()

Base:

The declarative base class used. By default, onegov.core.orm.Base is used.

Identity_secure:

True if the identity cookie is only transmitted over https. Only set this to False during development!

Identity_secret:

A random string used to sign the identity. By default a random string is generated. The drawback of this is the fact that users will be logged out every time the application restarts.

So provide your own if you don’t want that, but be sure to have a really long, really random key that you will never share with anyone!

Redis_url:

The redis url used (default is ‘redis://localhost:6379/0’).

File_storage:

The file_storage module to use. See https://docs.pyfilesystem.org/en/latest/filesystems.html

File_storage_options:

A dictionary of options passed to the __init__ method of the file_storage class.

The file storage is expected to work as is. For example, if fs.osfs.OSFS is used, the root_path is expected exist.

The file storage can be shared between different onegov.core applications. Each application automatically gets its own namespace inside this space.

Always_compile_theme:

If true, the theme is always compiled - no caching is employed.

Allow_shift_f5_comple:

If true, the theme is recompiled if shift+f5 is done on the browser (or shift + reload button click).

Csrf_secret:

A random string used to sign the csrf token. Make sure this differs from identity_secret! The algorithms behind identity_secret and the csrf protection differ. If the same secret is used we might leak information about said secret.

By default a random string is generated. The drawback of this is the fact that users won’t be able to submit their forms if the app is restarted in the background.

So provide your own, but be sure to have a really long, really random string that you will never share with anyone!

Csrf_time_limit:

The csrf time limit in seconds. Basically the amount of time a user has to submit a form, from the time it’s rendered.

Defaults to 1’200s (20 minutes).

Mail:

A dictionary keyed by e-mail category (i.e. ‘marketing’, ‘transactional’) with the following subkeys:

• host: The mail server to send e-mails from.

• port: The port used for the mail server.

• force_tls: True if TLS should be forced.

• username: The mail username

• password: The mail password

• sender: The mail sender

• use_directory: True if a mail directory should be used

• directory: Path to the directory that should be used

Mail_use_directory:

If true, mails are stored in the maildir defined through mail_directory. There, some other process is supposed to pick up the e-mails and send them.

Mail_directory:

The directory (maildir) where mails are stored if if mail_use_directory is set to True.

Sql_query_report:

Prints out a report sql queries for each request, unless False. Valid values are:

• ‘summary’ (only show the number of queries)

• ‘redundant’ (show summary and the actual redundant queries)

• ‘all’ (show summary and all executed queries)

Do not use in production!

Profile:

If true, profiles the request and stores the result in the profiles folder with the following format: YYYY-MM-DD hh:mm:ss.profile

Do not use in production!

Print_exceptions:

If true, exceptions are printed to stderr. Note that you should usually configure logging through onegov.server. This is mainly used for certain unit tests where we use WSGI more directly.

configure_dsn(*, dsn: str | None = None, base: type[Any] = Base, **cfg: Any) → None

configure_redis(*, redis_url: str = 'redis://127.0.0.1:6379/0', **cfg: Any) → None

configure_secrets(*, identity_secure: bool = True, identity_secret: str | None = None, csrf_secret: str | None = None, csrf_time_limit: float = 1200, **cfg: Any) → None

configure_yubikey(*, yubikey_client_id: str | None = None, yubikey_secret_key: str | None = None, **cfg: Any) → None

configure_mtan_second_factor(*, mtan_second_factor_enabled: bool = False, mtan_automatic_setup: bool = False, **cfg: Any) → None

configure_totp(*, totp_enabled: bool = True, **cfg: Any) → None

configure_filestorage(**cfg: Any) → None

configure_debug(*, always_compile_theme: bool = False, allow_shift_f5_compile: bool = False, sql_query_report: Literal[False, 'summary', 'redundant', 'all'] = False, profile: bool = False, print_exceptions: bool = False, **cfg: Any) → None

configure_mail(*, mail: dict[str, Any] | None = None, **cfg: Any) → None

configure_sms(*, sms_directory: str | None = None, sms: dict[str, Any] | None = None, **cfg: Any) → None

configure_hipchat(*, hipchat_token: str | None = None, hipchat_room_id: str | None = None, **cfg: Any) → None

configure_zulip(*, zulip_url: str | None = None, zulip_stream: str | None = None, zulip_user: str | None = None, zulip_key: str | None = None, **cfg: Any) → None

configure_content_security_policy(*, content_security_policy_enabled: bool = True, content_security_policy_report_uri: str | None = None, content_security_policy_report_only: bool = False, content_security_policy_report_sample_rate: float = 0.0, **cfg: Any) → None

configure_sentry(*, sentry_dsn: str | None = None, **cfg: Any) → None

property is_sentry_supported: bool

set_application_id(application_id: str) → None

Set before the request is handled. Gets the schema from the application id and makes sure it exists, if a database connection is present.

get_cache(name: str, expiration_time: float) → onegov.core.cache.RedisCacheRegion

Gets a cache bound to this application id.

property session_cache: onegov.core.cache.RedisCacheRegion

A cache that is kept for a long-ish time.

property cache: onegov.core.cache.RedisCacheRegion

A cache that might be invalidated frequently.

property settings: morepath.settings.SettingRegistry

Returns the settings bound to this app.

property application_id_hash: str

The application_id as hash, use this if the application_id can be read by the user -> this obfuscates things slightly.

object_by_path(path: str, with_view_name: Literal[False] = ...) → object | None

object_by_path(path: str, with_view_name: Literal[True]) → tuple[object | None, str | None]

Takes a path and returns the object associated with it. If a scheme or a host is passed it is ignored.

Be careful if you use this function with user provided urls, we load objects here, not views. Therefore no security restrictions apply.

The first use case of this function is to provide a generic copy/paste functionality. There, we only allow urls to be copied which have been previously signed by the server.

Safeguards like this are necessary if the user has the ability to somehow influence the path!

permission_by_view(model: type[object] | object, view_name: str | None = None) → type[core.security.permissions.Intent]

Returns the permission required for the given model and view_name.

The model may be an instance or a class.

If the view cannot be evaluated, a KeyError is raised.

property session: Callable[[], Session]

Alias for self.session_manager.session.

send_marketing_email(reply_to: Address | str | None = None, receivers: SequenceOrScalar[Address | str] = (), cc: SequenceOrScalar[Address | str] = (), bcc: SequenceOrScalar[Address | str] = (), subject: str | None = None, content: str | None = None, attachments: Iterable[Attachment | StrPath] = (), headers: dict[str, str] | None = None, plaintext: str | None = None) → None

Sends an e-mail categorised as marketing.

This includes but is not limited to:

• Announcements

• Newsletters

• Promotional E-Mails

When in doubt, send a marketing e-mail. Transactional e-mails are sacred and should only be used if necessary. This ensures that the important stuff is reaching our customers!

However, marketing emails will always need to contain an unsubscribe link in the email body and in a List-Unsubscribe header.

send_marketing_email_batch(prepared_emails: Iterable[EmailJsonDict]) → None

Sends an e-mail batch categorised as marketing.

This includes but is not limited to:

• Announcements

• Newsletters

• Promotional E-Mails

When in doubt, send a marketing e-mail. Transactional e-mails are sacred and should only be used if necessary. This ensures that the important stuff is reaching our customers!

However, marketing emails will always need to contain an unsubscribe link in the email body and in a List-Unsubscribe header.

Parameters:

prepared_emails – A list of emails prepared using app.prepare_email

Supplying anything other than stream=’marketing’ in prepare_email will be considered an error.

Batches will be split automatically according to API limits.

send_transactional_email(reply_to: Address | str | None = None, receivers: SequenceOrScalar[Address | str] = (), cc: SequenceOrScalar[Address | str] = (), bcc: SequenceOrScalar[Address | str] = (), subject: str | None = None, content: str | None = None, attachments: Iterable[Attachment | StrPath] = (), headers: dict[str, str] | None = None, plaintext: str | None = None) → None

Sends an e-mail categorised as transactional.

This is limited to:

• Welcome emails

• Reset passwords emails

• Notifications

• Weekly digests

• Receipts and invoices

send_transactional_email_batch(prepared_emails: Iterable[EmailJsonDict]) → None

Sends an e-mail categorised as transactional.

This is limited to:

• Welcome emails

• Reset passwords emails

• Notifications

• Weekly digests

• Receipts and invoices

Parameters:

prepared_emails – A list of emails prepared using app.prepare_email

Supplying anything other than stream=’transactional’ in prepare_email will be considered an error.

Batches will be split automatically according to API limits.

prepare_email(reply_to: Address | str | None = None, category: Literal['marketing', 'transactional'] = 'marketing', receivers: SequenceOrScalar[Address | str] = (), cc: SequenceOrScalar[Address | str] = (), bcc: SequenceOrScalar[Address | str] = (), subject: str | None = None, content: str | None = None, attachments: Iterable[Attachment | StrPath] = (), headers: dict[str, str] | None = None, plaintext: str | None = None) → core.types.EmailJsonDict

Common path for batch and single mail sending. Use this the same way you would use send_email then pass the prepared emails in a list or another iterable to the batch send method.

send_email(reply_to: Address | str | None = None, category: Literal['marketing', 'transactional'] = 'marketing', receivers: SequenceOrScalar[Address | str] = (), cc: SequenceOrScalar[Address | str] = (), bcc: SequenceOrScalar[Address | str] = (), subject: str | None = None, content: str | None = None, attachments: Iterable[Attachment | StrPath] = (), headers: dict[str, str] | None = None, plaintext: str | None = None) → None

Sends a plain-text e-mail to the given recipients. A reply to address is used to enable people to answer to the e-mail which is usually sent by a noreply kind of e-mail address.

E-mails sent through this method are bound to the current transaction. If that transaction is aborted or not commited, the e-mail is not sent.

Usually you’ll use this method inside a request, where transactions are automatically commited at the end.

send_email_batch(prepared_emails: Iterable[EmailJsonDict], category: Literal['marketing', 'transactional'] = 'marketing') → None

Sends an e-mail batch.

Parameters:

prepared_emails – A list of emails prepared using app.prepare_email

Batches will be split automatically according to API limits.

property can_deliver_sms: bool

Returns whether or not the current schema is configured for SMS delivery.

send_sms(receivers: SequenceOrScalar[str], content: str | bytes) → None

Sends an SMS by writing a file to the sms_directory of the principal.

receivers can be a single phone number or a collection of numbers. Delivery will be split into multiple batches if the number of receivers exceeds 1000, this is due to a limit in the ASPSMS API. This also means more than one file is written in such cases. They will share the same timestamp but will have a batch number prefixed.

SMS sent through this method are bound to the current transaction. If that transaction is aborted or not commited, the SMS is not sent.

Usually you’ll use this method inside a request, where transactions are automatically commited at the end.

send_zulip(subject: str, content: str) → onegov.core.utils.PostThread | None

Sends a zulip chat message asynchronously.

We are using the stream message method of zulip: https://zulipchat.com/api/stream-message

Returns the thread object to allow waiting by calling join.

property static_files: list[str]

A list of static_files paths registered through the onegov.core.directive.StaticDirectoryAction directive.

To register a static files path:

@App.static_directory()
def get_static_directory():
    return 'static'

For this to work, server_static_files has to be set to true.

When a child application registers a directory, the directory will be considered first, before falling back to the parent’s static directory.

property serve_static_files: bool

Returns True if /static files should be served. Needs to be enabled manually.

Note that even if the static files are not served, /static path is still served, it just won’t return anything but a 404.

Note also that static files are served publicly. You can override this in your application, but doing that and testing for it is on you!

See also: onegov.core.static.

application_bound_identity(userid: str, uid: str, groupid: str | None, role: str) → morepath.authentication.Identity

Returns a new morepath identity for the given userid, group and role, bound to this application.

property filestorage: SubFS[FS] | None

Returns a filestorage object bound to the current application. Based on this nifty module:

https://docs.pyfilesystem.org/en/latest/

The file storage returned is guaranteed to be independent of other applications (the scope is the application_id, not just the class).

There is no guarantee as to what file storage backend is actually used. It’s quite possible that the file storage will be somewhere online in the future (e.g. S3).

Therefore, the urls for the file storage should always be acquired through onegov.core.request.CoreRequest.filestorage_link().

The backend is configured through configure_application().

For a list of methods available on the resulting object, consult this list: https://docs.pyfilesystem.org/en/latest/interface.html.

If no filestorage is available, this returns None. See self.has_filestorage.

WARNING: Files stored in the filestorage are available publicly! All someone has to do is to guess the filename. To get an unguessable filename use onegov.core.filestorage.random_filename().

The reason for this is the fact that filestorage may be something external in the future.

This should not deter you from using this for user uploads, though you should be careful. If you want to be sure that your application stores files locally, use some other ways of storing those files.

Example:

from onegov.core import filestorage

filename = filestorage.random_filename()
app.filestorage.writetext(filename, 'Lorem Ipsum')

# returns either an url like '/files/4ec56cc005c594880a...'
# or maybe 'https://amazonaws.com/onegov-cloud/32746/220592/q...'
request.filestorage_link(filename)

property themestorage: SubFS[FS] | None

Returns a storage object meant for themes, shared by all applications.

Only use this for theming, nothing else!

property theme_options: dict[str, Any]

Returns the application-bound theme options.

property translations: dict[str, gettext.GNUTranslations]

Returns all available translations keyed by language.

property chameleon_translations: dict[str, translationstring._ChameleonTranslate]

Returns all available translations for chameleon.

property locales: set[str]

Returns all available locales in a set.

property default_locale: str | None

Returns the default locale.

property identity_secret: str

The identity secret, guaranteed to only be valid for the current application id.

property csrf_secret: str

The identity secret, guaranteed to only be valid for the current application id.

sign(text: str, salt: str = 'generic-signer') → str

Signs a text with the identity secret.

The text is signed together with the application id, so if one application signs a text another won’t be able to unsign it.

unsign(text: str, salt: str = 'generic-signer') → str | None

Unsigns a signed text, returning None if unsuccessful.

core.framework.get_webasset_url() → str

The webassets url needs to be unique so we can fix it before returning the generated html. See fix_webassets_url_factory().

core.framework.get_js_filter() → str

core.framework.get_css_filter() → str

core.framework.get_jsx_filter() → str

core.framework.fix_webassets_url_factory(app: Framework, handler: Callable[[CoreRequest], Response]) → Callable[[CoreRequest], Response]

core.framework.get_retry_attempts() → int

core.framework.get_cronjobs_enabled() → bool

If this value is set to False, all cronjobs are disabled. Only use this during testing. Cronjobs have no impact on your application, unless there are defined cronjobs, in which case they are there for a reason.

core.framework.default_content_security_policy() → more.content_security.ContentSecurityPolicy

The default content security policy used throughout OneGov.

core.framework.default_policy_apply_factory() → Callable[[ContentSecurityPolicy, CoreRequest, Response], None]

Adds the content security policy report settings from the yaml.

core.framework.http_conflict_tween_factory(app: Framework, handler: Callable[[CoreRequest], Response]) → Callable[[CoreRequest], Response]

core.framework.activate_session_manager_factory(app: Framework, handler: Callable[[CoreRequest], Response]) → Callable[[CoreRequest], Response]

Activate the session manager before each transaction.

core.framework.close_session_after_request_factory(app: Framework, handler: Callable[[CoreRequest], Response]) → Callable[[CoreRequest], Response]

Closes the session after each request.

This frees up connections that are unused, without costing us any request performance from what I can measure.

core.framework.current_language_tween_factory(app: Framework, handler: Callable[[CoreRequest], Response]) → Callable[[CoreRequest], Response]

core.framework.spawn_cronjob_thread_tween_factory(app: Framework, handler: Callable[[CoreRequest], Response]) → Callable[[CoreRequest], Response]