Stefan Scherfke - python

Publishing to PyPI with a Trusted Publisher from GitLab CI/CD

2024-11-14T07:18:00+01:00

PyPA’s Trusted Publishers let you upload Python packages directly from your CI pipeline to PyPI. And you don’t need any long-lived secrets like API tokens. This makes uploading Python packages not only easier than ever and more secure, too.

In this article, we’ll look at what Trusted Publishers are and how they’re more secure than using API tokens or a username/password combo. We’ll also learn how to set up our GitLab CI/CD pipeline to:

continuously test the release processes with the TestPyPI on every push to main,
automatically perform PyPI releases on every Git tag, and
additionally secure the process with GitLab (deployment) environments.

The official documentation explains most of this, but it doesn’t go into much depth regarding GitLab pipelines and leaves a few details unexplained.

Why should I want to use this?

API tokens aren’t inherently insecure, but they do have a few drawbacks:

If they are passed as environment variables, there’s a chance they’ll leak (think of a debug env | sort command in your pipeline).
If you don’t watch out, bad co-maintainers can steal the token and do mischief with it.
You have to manually renew the token from time to time, which can be annoying in the long run.

Trusted Publishers can avoid these problems or, at the very least, reduce their risk:

You don’t have to manually renew any long-lived tokens.
All tokens are short-lived. Even if they leak, they can’t be misused for long.

After we’ve learned how Trusted Publishers and protected GitLab environments work, we will take another look at security considerations.

How do Trusted Publishers work?

The basic idea of Trusted Publishers is quite simple:

In PyPI’s project settings, you add a Trusted Publisher and configure it with the GitLab URL of your project.
PyPI will then only accept package uploads if the uploader can prove that the upload comes from a CI pipeline of that project.

The technical process behind this is based on the OpenID Connect (OIDC) standard.

Essentially, the process works like this:

In your CI pipeline, you request an ID token for PyPI.
GitLab injects the short-lived token into your pipeline as a (masked) environment variable. It is cryptographically signed by GitLab and contains, among other things, your project’s path with namespace.
You use this token to authenticate with PyPI and request another token for the actual package upload.
This API token can now be used just like “normal” project-scoped API tokens.

The Trusted Publishers documentation explains this in more detail.

One problem remains, though: An ID token can be requested in any pipeline job and in any branch. Malicious contributors could sneak in a pipeline job and make a corrupted release.

This is where environments come in.

Environments

GitLab environments represent your deployed code in your infrastructure. Think of your code running in a container in your production or testing Kubernetes cluster; or your Python package living on PyPI. :-)

The most important feature of environments in this context is access control: You can protect environments, restricting deployments to them. For protected environments, you can define users or roles that are allowed to perform deployments and that must approve deployments. For example, you could restrict deployments (uploads to PyPI) to all maintainers of your project, but only after you yourself have approved each release.

To use an environment in your CI/CD pipeline, you need to add it to a job in the .gitlab-ci.yml.

If we also store the name of the environment in the PyPI deployment settings, only uploads from that environment will be allowed, i.e. only uploads that have been authorized by selected people.

Only maintainers can deploy to the release environment and only after Stefan approved it.

Security Considerations

The last two sections have already hinted at this: GitLab environments are only truly secure if you can protect them.

Let’s take a step back and consider what threats we’re trying to protect against, so that we’ll then be able to choose the right approach:

Random people doing a merge request for your project.
Contributors with the developer role committing directly into your project.
Co-maintainers with more permissions then a developer.
A Jia Tan which you trust even more than the other maintainers.

What can we do about it?

Code in other people’s forks doesn’t have access to your project’s CI variables nor can it request OIDC ID tokens in your project’s name. But you need to carefully review each MR!
Contributors with only developer permissions can still request ID tokens. If you cannot use protected environments, using an API token stored in a protected CI/CD variable is a more secure approach. You should also protect your main branch and all tags (using the * pattern), so that devleopers only have access to feature branches. You’ll find it under Settings → Repository → Protected branches/tags.
Protected CI/CD variables do not protect you from malicious maintainers, though. Even if you only allow yourself to create tags, other maintainers still have access to protected variables. Protected environments with only a selected set of approvers is the most secure approach.
If a very trusted co-maintainer becomes malicious, there’s very little you can do. Carefully review all commits and read the audit logs (Secure → Audit Events).

So that means for you:

If you are the only maintainer of a small open source project, just use a Trusted Publisher with (unprotected) environments.
If you belong to a larger project with multiple maintainers, consider applying for GitLab for Open Source and use a Trusted Publisher with a protected environment.
If there are multiple contributors and you don’t have access to protected environments, use an API token stored in a protected CI/CD variable and try only grant developer permissions to contributors.

Putting it all together

Configuring your GitLab project to use a trusted publisher involves three main steps:

Update your project’s publishing settings on PyPI and TestPyPI.
Update the CI/CD settings for your GitLab project.
Update your project’s pyproject.toml and .gitlab-ci.yml.

PyPI Settings

Tell PyPI to trust your GitLab CI pipelines.

Log in to PyPI and go to your account’s Publishing settings. Here, you can manage and add trusted publishers for your project.
Add a new trusted publisher for GitLab as shown in the screenshot below.

Enter your project’s namespace (your GitLab username or the name of your organization), the project name, the filename of your CI def (usually .gitlab-ci.yml).

Use release as the environment name!
Repeat the same steps for the TestPyPI, but use release-test as environment name.

Add a trusted publisher on PyPI.

GitLab CI/CD Settings

You need to create two environments and protect the one for production releases.

Open your project in GitLab, then go to Operate → Environments and click Create an environment to create the production environment:
- Title: release
- Description: PyPI releases (or whatever you want)
- External URL: https://pypi.org/project/{your-project}/ (the URL is displayed in a few places in GitLab and helps you to quickly navigate to your project on PyPI.)
Click Save.

Add an environment for your deployments in Gitlab.

Click New environment (in the top right corner) to create the test environment:
- Title: release-test
- Description: TestPyPI releases (or whatever you want)
- External URL: https://test.pypi.org/project/{your-project}/
Click Save.
If protected environments are available (see the note above), navigate to Settings → CI/CD and open the Protected environments section. Click the Protect an environment button.
- Select environment: release
- Allowed to deploy: Choose a role or user, e.g. Maintainers.
- Approvers: Choose a role or user, e.g. yourself.

Restrict who can deploy into the release environment (and thus, upload to PyPI).

Changes in Project Files

In order to be able to upload each commit to the TestPyPI, we need a different version for each build. To achieve this, we can use hatch-vcs, setuptools_scm, or similar.

In the following example, we are going to use hatchling with hatch-vcs as the build backend and uv for everything else.

We configure the build backend in our pyproject.toml as follows:

[build-system]
requires = ["hatchling", "hatch-vcs"]
build-backend = "hatchling.build"

[tool.hatch.version]
source = "vcs"
raw-options = { local_scheme = "no-local-version" }  # TestPyPI lacks support for this

[project]
dynamic = ["version"]

Now lets open our project’s .gitlab-ci.yml which we’ll edit during the next steps.

Hint

The snippets in the next steps only show fragments of the .gitlab-ci.yml. I’ll post the complete file at the end of the article.

We need at least a build and a deploy stage:

stages:
  - 'build'
  # - 'test'
  # - ...
  - 'deploy'

Python build tools usually put their artifacts (binary wheels and source distributions) into dist/. This directory needs to be added to your pipeline artifacts, so that these files are available in later pipeline jobs:
```
build:
  stage: 'build'
  script:
    - 'uv build --out-dir=dist'
  artifacts:
    paths:
      - 'dist/'
```

For our use-case, we need two release jobs: One that uploads to the TestPyPI on each push (release-test) and one that uploads to the PyPI in tag pipelines (release).

Since both jobs are nearly the same, we’ll also define an “abstract base job” .release-base which the other two extend.

.release-base:
  # Abstract base job for "release" jobs.
  # Extending jobs must define the following variables:
  # - PYPI_OIDC_AUD: Audience for the ID token that GitLab
  #   issues to the pipeline job
  # - PYPI_OIDC_URL: PyPI endpoint for retrieving a publish
  #   token with GitLab’s ID token
  # - UV_PUBLISH_URL: PyPI endpoint for the actual upload
  stage: 'deploy'
  id_tokens:
    PYPI_ID_TOKEN:
      aud: '$PYPI_OIDC_AUD'
  script:
    # Use the GitLab ID token to retrieve an API token from PyPI
    - >-
      resp="$(curl -X POST "${PYPI_OIDC_URL}" -d "{\"token\":\"${PYPI_ID_TOKEN}\"}")"
    # Parse the response and extract the token
    - >-
      publish_token="$(python -c "import json; print(json.load('${resp}')['token'])")"
    # Upload the files from "dist/"
    - 'uv publish --token "$publish_token"'
    # Print the link to PyPI so we can quickly go there to verify the result:
    - 'version="$(uv run --with hatch-vcs hatchling version)"'
    - 'echo -e "\033[34;1mPackage on PyPI:\033[0m ${CI_ENVIRONMENT_URL}${version}/"'

Now we can add the release-test job. It extends .release-base, defines variables for the base job, and rules for when the job should run:

release-test:
  extends: '.release-base'
  rules:
    # Only run if it's a pipeline for the default branch or a tag:
    - if: '$CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH || $CI_COMMIT_TAG'
  environment:
    name: 'release-test'
    url: 'https://test.pypi.org/project/typed-settings/'
  variables:
    PYPI_OIDC_AUD: 'testpypi'
    PYPI_OIDC_URL: 'https://test.pypi.org/_/oidc/mint-token'
    UV_PUBLISH_URL: 'https://test.pypi.org/legacy/'

The release job looks very similar, but the variables have different values and the job only runs in tag pipelines.

release:
  extends: '.release-base'
  rules:
    # Only run in tag pipelines:
    - if: '$CI_COMMIT_TAG'
  environment:
    name: 'release'
    url: 'https://pypi.org/project/typed-settings/'
  variables:
    PYPI_OIDC_AUD: 'pypi'
    PYPI_OIDC_URL: 'https://pypi.org/_/oidc/mint-token'
    UV_PUBLISH_URL: 'https://upload.pypi.org/legacy/'

The output of the release will look like this. There’s also a link that takes you directly to the release on PyPI.

That’s it. You should now be able to automatically create PyPI releases directly from your GitLab CI/CD pipeline. 🎉

A successful GitLab CI/CD pipeline for Typed Settings’ v24.6.0 release.

If you run into any problems, you can

check if the settings on PyPI match your GitLab project,
read the Trusted Publishers docs,
read the GitLAB CI/CD YAML syntax reference,
read the docs for GitLab environments and GitLab OIDC authentication.

You can leave comments over at Mastodon or Bluesky.

And, as promised, here is the complete (but still minimal) .gitlab-ci.yml from the snippets above. If you want to see a real-world example, you can take a look at Typed Settings pipeline definition.

# .gitlab-ci.yml

stages:
  - 'build'
  # - 'test'
  # - ...
  - 'deploy'

build:
  stage: 'build'
  script:
    - 'uv build --out-dir=dist'
  artifacts:
    paths:
      - 'dist/'

.release-base:
  # Abstract base job for "release" jobs.
  # Extending jobs must define the following variables:
  # - PYPI_OIDC_AUD: Audience for the ID token that GitLab issues to the pipeline job
  # - PYPI_OIDC_URL: PyPI endpoint for retrieving a publish token with GitLab’s ID token
  # - UV_PUBLISH_URL: PyPI endpoint for the actual upload
  stage: 'deploy'
  id_tokens:
    PYPI_ID_TOKEN:
      aud: '$PYPI_OIDC_AUD'
  script:
    - >-
      resp="$(curl -X POST "${PYPI_OIDC_URL}" -d "{\"token\":\"${PYPI_ID_TOKEN}\"}")"
    - >-
      publish_token="$(python -c "import json; print(json.load('${resp}')['token'])")"
    - 'uv publish --token "$publish_token"'
    - 'version="$(uv run --with hatch-vcs hatchling version)"'
    - 'echo -e "\033[34;1mPackage on PyPI:\033[0m ${CI_ENVIRONMENT_URL}${version}/"'

release-test:
  extends: '.release-base'
  rules:
    - if: '$CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH || $CI_COMMIT_TAG'
  environment:
    name: 'release-test'
    url: 'https://test.pypi.org/project/typed-settings/'
  variables:
    PYPI_OIDC_AUD: 'testpypi'
    PYPI_OIDC_URL: 'https://test.pypi.org/_/oidc/mint-token'
    UV_PUBLISH_URL: 'https://test.pypi.org/legacy/'

release:
  extends: '.release-base'
  rules:
    - if: '$CI_COMMIT_TAG'
  environment:
    name: 'release'
    url: 'https://pypi.org/project/typed-settings/'
  variables:
    PYPI_OIDC_AUD: 'pypi'
    PYPI_OIDC_URL: 'https://pypi.org/_/oidc/mint-token'
    UV_PUBLISH_URL: 'https://upload.pypi.org/legacy/'

Typed Settings

2020-11-30T22:14:00+01:00

There are already several settings libraries like Dynaconf, Environ Config, or Pydantic – just to name a few. I have written a new one: Typed Settings.

What makes it different?

Settings are defined as type-hinted, immutable (frozen) attrs classes. Values are automatically converted to the proper type when they are loaded. Apart from simple data types, Typed Settings supports datetimes, enums, nested attrs classes and various container types (like lists). The auto-converter can be extended to handle additional types.

Settings can be loaded from multiple config files. Config files can contain settings for multiple apps (like pyproject.toml). Different deployment environments use different config files (this is in contrast to Dynaconf, where a single config file specifies the settings for all environments in different sections). Currently, only TOML is supported. Support for YAML or .env may follow later.

Value interpolation as in Dynaconf is not yet supported, but planned.

Search paths for config files have to be explicitly stated – either directly in the app or via an environment variable.

Environment variables can also be used to override settings. The prefix is customizable and this feature can also be disabled.

Finally, Typed Settings can generate Click options for command line applications. You CLI function will receive all options nicely packed together as a single instance of your settings class.

Specialized secrets stores like HashiCorp Vault are not (yet?) supported.

Invalid values or undefined options in config files raise an error instead of being silently ignored. Config files can optionally be marked as mandatory and an error will be raised if such a file cannot be found.

To aid with debugging, Typed Settings uses Python’s logging module to log config files that are being loaded (or that cannot be found) as well as looked up env vars.

Everything is thoroughly tested, the test coverage is at 100%.

An Example

Here is a very simple example that demonstrates how you can load settings from a statically defined config file and from environment variables:

# example.py
import typed_settings as ts

@ts.settings
class Settings:
    option_one: str
    option_two: int

settings = ts.load_settings(
    cls=Settings,
    appname="example",
    config_files=["settings.py"],  # Paths can also be set via env var
)
print(settings)

# settings.toml
[example]
option_one = "value"

$ EXAMPLE_OPTION_TWO=2 python example.py
Settings(option_one="value", option_two=2)

The README and documentation contain more examples.

Project Status

The recently released version 0.9 contains all features that are planed for version 1.0.0. Additional features are already on the roadmap.

What’s missing for the first stable release is mainly documentation as well as more real life testing. I already use Typed Settings for a few projects in our company and will perspectively try to replace our old settings system with it.

Raise … from … in Python

2020-10-28T11:23:00+01:00

When you recently upgraded to pylint 2.6.0, you may have stumbled across a new warning:

src/mylib/core.py:74:20: W0707:
  Consider explicitly re-raising using the
  'from' keyword (raise-missing-from)

The reason for this message is an exception that you raised from within an except block like this:

>>> class MyLibError(Exception):
...     """Base class for all errors raised by mylib"""
...
>>> def do_stuff("onoes"):
...     try:
...         int(text)
...     except ValueError as e:
...         raise MyLibError(e)

When you run do_stuff(), you’ll get the following traceback:

>>> do_stuff(text)
Traceback (most recent call last):
  File "<stdin>", line 3, in do_stuff
ValueError: invalid literal for int() with base 10: 'onoes'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 5, in do_stuff
__main__.MyLibError: invalid literal for int() with base 10: 'onoes'

The important line here is:

During handling of the above exception, another exception occurred

This means that while you were handling the ValueError, another (unexpected) exception occurred: a MyLibError.

But this is not what we wanted to do – we wanted to replace the ValueError with MyLibError, so that our uses only have to handle a single exception type!

Enter `raise … from …`

To express I want to modify and forward an existing exception, you can use the raise NewException from cause syntax:

>>> def do_stuff(text):
...     try:
...         int(text)
...     except ValueError as e:
...         raise MyLibError(e) from e

When we run this now, we’ll get a different traceback printed:

>>> do_stuff("onoes")
Traceback (most recent call last):
  File "<stdin>", line 3, in do_stuff
ValueError: invalid literal for int() with base 10: 'onoes'

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 5, in do_stuff
__main__.MyLibError: invalid literal for int() with base 10: 'onoes'

Your users will now receive a MyLibError with the attached information, that the cause of this error was a ValueError somewhere in your code.

When the underlying cause is not important

If your users shouldn’t care about the underlying cause, because the new exception contains all the relevant information (i.e., that the provided input cannot be parsed), you may also omit the cause:

>>> def do_stuff(text):
...     try:
...         int(text)
...     except ValueError as e:
...         raise MyLibError(e) from None

When you run this now, you’ll get a nice an clean traceback:

>>> do_stuff("onoes")
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 5, in do_stuff
__main__.MyLibError: invalid literal for int() with base 10: 'onoes'

Summary

When you raise an exception from within an except block in Python, you have three options:

If a new/unexpected exception occurs in the code handling the original exception, raise NewException.
If you want to wrap the original exception(s) (e.g., with a common base exception to reduce complexity for your users), raise NewException from cause.
If you want to hide the original exception because it is irrelevant for your users, raise NewException from None.

Attrs, Dataclasses and Pydantic

2020-05-29T09:36:00+02:00

I’ve been using attrs for a long time now and I am really liking it. It is very flexible, has a nice API, is well documented and maintained, and has no runtime requirements.

The main idea behind attrs was to make writing classes with lots of data attributes (“data classes”) easier. Instead of this:

class Data:
    def __init__(self, spam, eggs):
        self.spam = spam
        self.eggs = eggs

It lets you write this:

@attr.s
class Data:
    spam = attr.ib()
    eggs = attr.ib()

Attrs also adds a nice string representation, comparison methods, optional validation and lots of other stuff to your classes, if you want to. You can also opt out of everything; attrs is very flexible.

Attrs became so popular, that since Python 3.7 we also have the dataclasses module in the standard library. It is predominantly inspired by attrs (the attrs team was involved in the design of data classes) but has a smaller feature set and will evolve a lot slower. But you can use it out-of-the box without adding a new requirement to your package.

Pydantic’s development roughly started during Python 3.7 development, too. Its main focus is on data validation, settings management and JSON (de)serialisation, therefore it is located at a higher level ob abstraction. Out of the box, it will recursively validate and convert all data that you pour into your model:

>>> from datetime import datetime
>>> from pydantic import BaseModel
>>>
>>> class Child(BaseModel):
...     x: int
...     y: int
...     d: datetime
...
>>> class Parent(BaseModel):
...     name: str
...     child: Child
...
>>> data = {
...     'name': 'spam',
...     'child': {
...         'x': 23,
...         'y': '42',  # sic!
...         'd': '2020-05-04T13:37:00',
...     },
... }
>>> Parent(**data)
Parent(name='spam', child=Child(x=23, y=42, d=datetime.datetime(2020, 5, 4, 13, 37)))

I only learned about pydantic when I started to work with FastAPI. FastAPI is a fast, asynchronous web framework specifically designed for building REST APIs. It uses pydantic for schema definition and data validation.

Since then, I asked myself: Why not attrs? What’s the benefit of pydantic over the widely used and mature attrs? Can or should it replace attrs?

As I begin to write this article, I still don’t know the answer to these questions. So lets explore attrs, data classes and pydantic!

Simple class definition

Originally, attrs classes were created by using the @attr.s() (or @attr.attrs() class decorator. Fields had to be created via the attr.ib() (or @attr.attrib()) factory function. By now, you can also create them nearly like data classes.

The recommended way for creating pydantic models is to subclass pydantic.BaseModel. This means that in contrast to data classes, all models inherit some “public” methods (e.g., for JSON serialization) which you need to be aware of. However, pydantic allows you to create stdlib data classes extended with validation, too.

Here are some very simple examples for data classes / models:

>>> import attr
>>> import dataclasses
>>> import pydantic
...
...
>>> # Simple data classes are supported by all libraries:
>>> @attr.dataclass
... # @dataclasses.dataclass
... # @pydantic.dataclasses.dataclass
... class Data:
...     name: str
...     value: float
...
>>> Data('Spam', 3.14)
Data(name='Spam', value=3.14)
...
...
>>> @attr.s
... class Data:
...     name = attr.ib()
...     value = attr.ib()
...
>>> Data('Spam', 3.14)
Data(name='Spam', value=3.14)
...
...
>>> class Data(pydantic.BaseModel):
...     name: str
...     value: float
...
>>> Data(name='Spam', value=3.14)
Data(name='Spam', value=3.14)

Pydantic models enforce keyword only arguments when creating new instances. This is a bit tedious for classes with only a few attributes but with larger models, you’re likely going to use keyword arguments anyways. The benefit of kw-only arguments is, that it doesn’t matter if you list attributes with a default before ones without a default.

Data classes support positional as well as keyword arguments. Passing values by position is very convenient for smaller classes but that also means that you must define all fields without a default value first and the ones with a default value afterwards. This may prevent you from grouping similar attributes, when only some of them have a default value.

Attrs supports both ways. The default is to allow positional and keyword arguments like data classes. You can enable kw-only arguments by passing kw_only=True to the class decorator.

Another major difference is that Pydantic always validates and converts all attribute values, but more on that later.

Class and attribute customistaion

All three libraries let you customize the created fields as well as the class itself.

In data classes and attrs, you can customize your class by passing additional arguments to the class decorator. Pydantic models can define a nested Config class for the same purpose.

Attributes can be customized via special factory functions. Instead of specifying an attribute like this: name: type [= default], you you do: name: type = field_factory(). This function is named [attr.]ib()/attrib() in attrs, field() with data classes and Field() in pydantic.

Using these functions, you can specify default values, validators, meta data and other attributes. The following tables let you compare the customisation features that each library provides:

Attribute / field settings
	attrs	data classes	pydantic
Explicit no default	`NOTHING` ¹	`MISSING` ¹	`...` ¹
Default factory	yes	yes	yes
Validators	yes ²	no	no ^2,3
Constraints	no	no	const, regex, length, number range, …
Converters	yes ²	no	no ^2,3
Exclude field from	repr, eq, order, hash, init	repr, compare, hash, init	ø
Add arbitrary metadata	yes	yes	yes
Additional docs	no	no	title, description

Class customisation and instantiation
	attrs	data classes	pydantic
Auto create methods for	str, repr, equality, ordering, hash, init	repr, equality, ordering, hash, init	str, repr, equality, init
Keyword args only	optional	no	yes
Faux immutability / Freezing	yes	yes	yes
Slots	yes	no	no
Safe to subclass exceptions	yes	no	no
Dynamic creation	yes	yes	yes
Instantiate from dict	yes	yes	yes, recursively
Instantiate from objects	no	no	optional, recursively
Instantiate from JSON	no	no	yes, recursively
Instantiate from env. vars.	no	no	yes, recursively

Generated methods

All libraries create useful “dunder” methods (like __init__() or __str__()). Attrs can generate the most methods, followed by data classes and Pydantic. Attrs and data classes also allow you to selectively disable the generation of certain methods.

Attrs is the only library that generates __slots__ and is also the only one that has explicit support for subclassing exceptions.

Default values

Without a field factory, default values for fields are simply assigned to the field name, e.g., value: int = 42. When you use a field factory, you can/need to pass a default value as argument to that function. In pydantic, the default value is always passed as first positional argument. In order to express “this attribute has no default”, you use the elipsis literal (...). Data classes use the optional keyword argument default instead. Attrs lets you choose - you can pass a default value by position or as keyword argument.

Another difference is that pydantic allows you to use mutable objects like lists or dicts as default values. Attrs and data classes prohibit this for good reason. To prevent bugs with mutable defaults, pydantic deep-copies the default value for each new instance.

You you can specify factories for default values with all libraries.

Freezing and functional programming

You can create pseudo immutable classes with all libraries. Immutable/frozen instances prevent you from changing attribute values. This helps when you aim to program in a more functional style. However, if attributes themselves are mutable (like lists or dicts), you can still change these!

In attrs and data classes, you pass frozen=True to the class decorator. In pydantic, you set allow_mutation = False in the nested Config class.

Attrs and data classes only generate dunder protocol methods, so your classes are “clean”. Having struct-like, frozen instances make it relatively easy to write purely functional code, that can be more robust and easier to test than code with a lot of side effects.

Pydantic models, on the other hand, use inheritance and always have some methods, e.g., for converting an instance from or to JSON. This facilitates a more object-orient programming style, which can be a bit more convenient in some situations.

Instantiation, validation and conversion

The main differentiating features of pydantic are its abilities to create, validate and serialize classes.

You can instantiate pydantic models not only from dicts/keyword arguments but also from other data classes (ORM mode), from environment variables, and raw JSON. Pydantic will then not only validate/convert basic data types but also more advanced types like datetimes. On top of that, it will recursively create nested model instances, as shown in the example above.

Model instances can directly be exported to dicts and JSON via the .dict()/.json() methods.

To achieve something similar in attrs or data classes, you need to install an extension package like, for example, cattrs. And even then, Pydantic has a far better user experience.

Apart from that, all libraries allow you to define custom validator and converter functions. You can either pass these functions to the field factories or define decorated methods in your class.

Metadata and schema generation

Pydantic can not only serialize model instances but also the schema of the model classes themselves. This is, for example, used by FastAPI to generate the OpenAPI spec for an API.

To aid the documentation of the generated schemas, every field can have a title and a description attribute. These are not used for docstrings, though.

Documentation

In a way, the documentation of all three projects mirrors their feature sets.

It is of high quality in all cases, but technically and in terms of content very different.

The data classes documentation is part of Python’s stdlib documentation and the briefest of all candidates, but it covers everything you need to know. It contains a direct link to the source code that also has many helpful comments.

The attrs docs contain example based guides, in-depth discussions of certain features and design decisions as well as an exhaustive API reference. It uses Sphinx and, for the API reference, the autodoc extension. It provides an objects inventory which allows you to cross-reference attrs classes and functions from your own documentation via intersphinx.

The pydantic documentation is also very well written and contains many good examples that explain almost all functionality in great detail. However, it follows the unfortunate trend of using MkDocs as a documentation system. I assume that this is easier to set-up then Sphinx and allows you to use Markdown instead of ReStructuredText, but it is also lacking lots of important features and I also don’t like its UX. It has two navigation menus – one on the left for whole document’s TOC and one on the right for the current page. More serious, however, is the absence of an API reference. There is also no cross referencing (e.g., links from class and function names to their section in the API reference) and thus no objects inventory that can be used for inter-project cross referencing via Sphinx’ intersphinx extension. Even pydantic’s source code barely includes any docstrings or other comments. This can be a hindrance when you want to solve more advanced problems.

Performance

For most use cases, the performance of a data classes library can be neglected. Performance differences only become noticeable when you create thousands or even millions of instances in a short amount of time.

However, the pydantic docs contain some benchmarks that suggest that pydantic is slightly ahead of attrs + cattrs in mean validation time. I was curious why pydantic, despite its larger feature set, was so fast, so I made my own benchmarks.

I briefly evaluate the attrs extension packages. The only one that offers reasonably convenient means for input validation and (de)serialization as well as good performance is cattrs.

I created benchmarks for three different use cases:

Simple classes and no need for extensive validation
Deserialization and validation of (more) complex (nested) classes
Serialization of complex (nested) classes

I calculated the time and memory consumption for handling 1 million instances of different variants of attrs and data classes as well as pydantic models.

Unsurprisingly, attrs and data classes are much faster than pydantic when no validation is needed. They also use a lot less memory.

I was expecting that the results would be much closer when it comes to validation/conversion and serialization, but even there, pydantic was a lot slower than attrs + cattrs.

I wondered why my benchmarks were so clearly in favor of attrs when the pydantic docs state that it is 1.4x faster than attrs + cattrs. I tried running the pydantic benchmarks myself and indeed I could reproduce these results. Wondering, why the results differed so much, I took a closer look at the benchmark’s source code. It turned out that the attrs + cattrs example used python-dateutil for parsing datetimes while pydantic uses its own implementation. I replaced datetuil.parser.parse() with the stdlib datetuil.fromisoformat() and the attrs + cattrs example suddenly became 6–7 times faster. (Note: fromisoformat() is not a general purpose parser!)

In defense of pydantic: The attrs + cattrs (de)serializers were specifically designed and implemented for this benchmark while Pydantic ships everything out-of-the box. Pydantic’s UX for these use cases is also more pleasant than that of attrs + cattrs.

You can find the source of all benchmark in the accompanying repository.

Summary

Attrs, data classes and pydantic seem very similar on a first glance, but they are very different when you take a closer look.

All three projects are of high quality, well documented and generally pleasant to use. Furthermore, they are different enough that each of them has its niche where it really shines.

The stdlib’s data classes module provides all you need for simple use cases and it does not add a new requirement to your project. Since it does not do any data validation, it is also quite fast.

When you need more features (more control over the generated class or data validation and conversion), you should use attrs. It is as fast as data classes are but its memory footprint is even smaller when you enable __slots__.

If you need extended input validation and data conversion, Pydantic is the tool of choice. The price of its many features and nice UX is a comparatively bad performance, though.

If you want to cut back on UX instead, the combination of attrs and cattrs might also be an alternative.

You can take a look at the benchmarks to get a feel for how the libraries can be used for different use cases and how they differ form each other.

I myself will stay with attrs as long as it can provide what I need. Otherwise I’ll use Pydantic.

Epilogue

I wish there was a library like mattrs (magic attrs) that combined Pydantic’s (de)serialization UX with attrs’ niceness and performance:

>>> from datetime import datetime
>>> from mattrs import dataclass, field, asjson
>>>
>>> @dataclass()
... class Child:
...     x: int = field(ge=0, le=100)
...     y: int = field(ge=0, le=100)
...     d: datetime
...
>>> @dataclass()
... class Parent:
...     name: str = field(re=r'^[a-z0-9-]+$')
...     child: Child
...
>>> data = {'name': 'spam', 'child': {'x': 23, 'y': '42', 'd': '2020-05-04T13:37:00'}}
>>> Parent(**data)
Parent(name='spam', child=Child(x=23, y=42, d=datetime.datetime(2020, 5, 4, 13, 37)))
>>> asjson(_)
{"name": "spam", "child": {"x": 23, "y": 42, "d": "2020-05-04T13:37:00"}}

Maybe it’s time for another side project? 🙊

Packaging Python inside your organization with GitLab and Conda

2019-04-18T13:37:00+02:00

Python Packaging has recently been discussed a lot, but the articles usually only focus on publishing (open source) code to PyPI.

But what do you do when your organization uses Python for in-house development and you can’t (or don’t want to) make everything Open Source? Where do you store and manage your code? How do you distribute your packages?

In this article, I describe how we solve this problem with GitLab, Conda and a few other tools.

You can find all code and examples referenced in this article under gitlab.com/ownconda. These tools and examples are using the own prefix in order to make a clear distinction between our own and third-party code. I will not necessarily update and fix the code, but it is released under the Blue Oak license so you can copy and use it. Any feedback is welcome, nonetheless.

Software selection

In this section I’ll briefly explain the reasons why we are using GitLab and Conda.

Code and issue management

Though you could use private repositories from one of the well-known cloud services, you should probably use a self-hosted service to retain full control over your code. In some countries it may even be forbidden to use a US cloud service for your organization’s data.

There are plenty of competitors in this field: GitLab, Gitea, Gogs, Gitbucket or Kallithea—just to name a few.

Our most important requirements are:

Repository management
Pull/Merge requests
Issue management
CI/CD pipelines

The only tool that (currently) meets these requirements is GitLab. It has a lot more features that are very useful for an organization wide use, e.g., LDAP and Kerberos support, issue labels and boards, Mattermost integration or Git LFS support. And—more importantly—it also has a really nice UX and is one of the few pieces of software that I actually enjoy using.

GitLab has a free core and some paid versions that add more features and support.

The package manager: Pip or Conda?

Pip is the official package installer for Python. It supports Python source distributions and (binary) Wheel packages. Pip only installs files in the current environment’s site-packages directory and can optionally create entry points in its bin directory. You can use Virtualenv to isolate different projects from another, and Devpi to host your own package index. Devpi can both, mirror/cache PyPI and store your own packages. The Python packaging ecosystem is overlooked by the Python Packaging Authority working group (PyPA).

Conda stems from the scientific community and is being developed by Anaconda. In contrast to Pip, Conda is a full-fledged package manager similar to apt or dnf. Like virtualenv, Conda can create isolated virtual environments. Conda is not directly compatible with Python’s setup.py or pyproject.toml files. Instead, you have to create a Conda recipe for every package and build it with conda-build. This is a bit more involved because you have to convert every package that you find on PyPI, but it also lets you patch and extend every package. With very little effort you can create a self-extracting Python distribution with a selection of custom packages (similar to the Miniconda distribution).

Conda-forge is a (relatively) new project that has a huge library of Conda recipes and packages. However, if you want full control over your own packages you may want to host and build everything on your own.

What to use?

Both, Conda and pip, allow you to host your own packages as well as 3rd party packages inside your organization.
Both, Conda and pip, provide isolated virtual environments.
Conda can package anything (Python, C-libraries, Rust apps, …) while Pip is exclusively for Python packages.
With Conda, you need to package and build everything on your own. Even packages from PyPI need to be re-packaged. On the other side, this makes it easier to patch and extend the package’s source.
Newer Conda versions allow you to build everything on your own, even GCC and libc. This is, however, not required and you can rely on some low-level system libraries like the manylinux standard for Wheels does. (You just have to decide which ones, but more on that later.)
Due to its larger scope, Conda is slower and more complex than Pip. In the past, even patch releases introduced backwards incompatible changes and bugs that broke our stack. However, the devs are very friendly and usually fix critical bugs quite fast. And maybe we would have had similar problems, too, if we used a Pip based stack.

Because we need to package more than just Python, we chose to use Conda. This dates back to at least to Conda v2.1 which was released in 2013. At that time, projects like conda-forge weren’t even in sight.

Supplementary tools

To aid our work with GitLab and Conda, we developed some supplementary tools. I have released a slightly modified version of them, called ownconda tools, alongside with this article.

The ownconda tools are a click based collection of commands that reside under the entry point ownconda.

Initially, they were only meant to help with the management of recipes for external packages, and with running the build/test/upload steps in our GitLab pipeline. But they have become a lot more powerful by now and even include a GitLab Runner that lets you run your projects’ pipelines locally (including artifacts handling, which the official gitlab-runner cannot do locally).

$ ownconda --help
Usage: ownconda [OPTIONS] COMMAND [ARGS]...

  Support tools for local development, CI/CD and Conda packaging.

Options:
  --help  Show this message and exit.

Commands:
  build                 Build all recipes in RECIPE_ROOT in the correct...
  check-for-updates     Update check for external packages in RECIPE_ROOT.
  ci                    Run a GitLab CI pipeline locally.
  completion            Print Bash or ZSH completion activation script.
  dep-graph             Create a dependency graph from a number of Conda...
  develop               Install PATHS in develop/editable mode.
  gitlab                Run a task on a number of GitLab projects.
  lint                  Run pylint for PATHS.
  make-docs             Run sphinx-build and upload generated html...
  prune-index           Delete old packages from the local Conda index at...
  pylintrc              Print the built-in pylintrc to stdout.
  pypi-recipe           Create or update recipes for PyPI packages.
  sec-check             Run some security checks for PATHS.
  show-updated-recipes  Show updated recipes in RECIPE_ROOT.
  test                  Run tests in PATHS.
  update-recipes        Update Conda recipes in RECIPE_ROOT.
  upload                Upload Conda packages in PKG_DIR.
  validate-recipes      Check if recipes in RECIPE_ROOT are valid.

I will talk about the various subcommands in more detail in later sections.

How it should work

The subject of packaging consists of several components: The platforms on which your code needs to build and run, the package manager and repository, management of external and internal packages, a custom Python distribution, and means to keep an overview over all packages and their dependencies. I will go into detail about each aspect in the following sections.

Runtime and build environment

Our packages need to run on Fedora desktop systems and on Centos 7. Packages built on Centos also run on Fedora, so we only have a single build environment: Centos 7.

We use different Docker images for our build pipeline and some deployments. The most important ones are centos7-ownconda-runtime and centos7-ownconda-develop. The former only contains a minimal setup to install and run Conda packages while the latter includes all build dependencies, conda-build and the ownconda tools.

If your OS landscape is more heterogeneous, you may need to add more build environments which makes things a bit more complicated—especially if you need to support macOS or even Windows.

To build Docker images in our GitLab pipelines, we use docker-in-docker. That means that the GitLab runners start docker containers that can access /var/run/dockers.sock to run docker build.

GitLab provides a Docker registry that allows any project to host its own images. However, if a project is private, other project’s pipelines can not access these images. For this reason, we have decided to serve Docker images from a separate host.

3rd party packages

We re-package all external dependencies as Conda packages and host them in our own Conda repository.

This has several benefits:

We can prohibit installing Software from other sources than our internal Conda repository.
If users want to depend on new libraries, we can propose alternatives that we might already have on our index. This keeps our tree of dependencies a bit smaller.
We cannot accidentally depend on packages with “bad” licenses.
We can add patches to fix bugs or extend the functionality of a package (e.g., we added our internal root certificate to Certifi).
We can reduce network traffic to external servers and are less dependent on their availability.

Recipe organization

We can either put the recipe for every package into its own repository (which is what conda-forge does) or use a single repository for all recipes (which is what we are doing).

The multi-repository approach makes it easier to only build packages that have changed. It also makes it easier to manage access levels if you have a lot of contributors that each only manage a few packages.

The single-repository approach has less overhead if you only have a few maintainers that take care of all the recipes. To identify updated packages that need re-building, we can use ownconda’s show-updated-recipes command.

Linking against system packages

With Conda, we can (and must) decide whether we want to link against system packages (e.g., installed with yum or use other Conda packages to satisfy a package’s dependencies.

One extreme would be to only build Python packages on our own and completely depend on system packages for all C libraries. The other extreme would be to build everything on our own, even glibc and gcc.

The former has a lot less overhead but becomes the more fragile the more heterogeneous your runtime environments become. The latter is a lot more complicated and involved but gives you more control and reliability.

We decided to take the middle ground between these two extremes: We build many libraries on our own but rely on the system’s gcc, glibc, and X11 libraries. This is quite similar to what the manylinux standard for Python Wheels does.

Recipes must list the system libraries that they link against. The rules for valid system libraries are encoded in ownconda validate-recipes and enforced by conda-build’s –error-overlinking option.

Recipe management

Recipes for Python packages can easily be created with ownconda pypi-recipe. This is similar to conda skeleton pypi but tailored to our needs. Recipes for other packages have to be created manually.

We also implemented an update check for our recipes. Every recipe contains a script called update_check.py which uses one of the update checkers provided by the ownconda tools.

These checkers can query PyPI, GitHub release lists and (FTP) directory listings, or crawl an entire website. The command ownconda check-for-updates runs the update scripts and compares the version numbers they find against the recipes’ current versions. It can also print URLs to the packages’ changelogs:

$ own check-for-updates --verbose .
  [████████████████████████████████████]  100%
Package: latest version (current version)
freetype 2.10.0 (2.9.1):
  https://www.freetype.org/index.html#news

python-attrs 19.1.0 (18.2.0):
  http://www.attrs.org/en/stable/changelog.html

python-certifi 2019.3.9 (2018.11.29):
  https://github.com/certifi/python-certifi/commits/master

...

qt5 5.12.2 (5.12.1):
  https://wiki.qt.io/Qt_5.12.2_Change_Files

readline 8.0.0 (7.0.5):
  https://tiswww.case.edu/php/chet/readline/CHANGES

We can then update all recipes with ownconda update-recipes:

$ ownconda update-recipes python-attrs ...
python-attrs
cd /data/ssd/home/stefan/Projects/ownconda/external-recipes && /home/stefan/ownconda/bin/python -m own_conda_tools pypi-recipe attrs -u
diff --git a/python-attrs/meta.yaml b/python-attrs/meta.yaml
index 7d167a8..9b3ea20 100644
--- a/python-attrs/meta.yaml
+++ b/python-attrs/meta.yaml
@@ -1,10 +1,10 @@
 package:
  name: attrs
-  version: 18.2.0
+  version: 19.1.0

 source:
-  url: https://files.pythonhosted.org/packages/0f/9e/26b1d194aab960063b266170e53c39f73ea0d0d3f5ce23313e0ec8ee9bdf/attrs-18.2.0.tar.gz
-  sha256: 10cbf6e27dbce8c30807caf056c8eb50917e0eaafe86347671b57254006c3e69
+  url: https://files.pythonhosted.org/packages/cc/d9/931a24cc5394f19383fbbe3e1147a0291276afa43a0dc3ed0d6cd9fda813/attrs-19.1.0.tar.gz
+  sha256: f0b870f674851ecbfbbbd364d6b5cbdff9dcedbc7f3f5e18a6891057f21fe399

 build:
-  number: 1
+  number: 0

...

The update process

Our Conda repository has various channels for packages of different maturity, e.g. experimental, testing, staging, and stable.

Updates are first built locally and uploaded to the testing channel for some manual testing.

If everything goes well, the updates are committed into the develop branch, pushed to GitLab and uploaded to the staging channel. We also send a changelog around to notify everyone about important updates and when they will be uploaded into the stable channel.

After a few days in testing, the updates are merged into the master branch and upload to the stable channel for production use.

This is a relatively save procedure which (usually) catches any problems before they go into production.

Example recipes

You can find the recipes for all packages required to run the ownconda tools here. As a bonus, I also added the recipes for NumPy and PyQt5.

Internal projects

Internal packages are structured in a similar way to most projects that you see on PyPI. We put the source code into src, the pytest tests into tests and the Sphinx docs into docs. We do not use namespace packages. They can lead to various nasty bugs. Instead, we just prefix all packages with own_ to avoid name clashes with other packages and to easily tell internal and external packages apart.

A project usually contains at least these files and directories.

The biggest difference to “normal” Python projects is the additional Conda recipe in each project. It contains all meta data and the requirements. The setup.py contains only the minimum amount of information to get the package installed via pip:

Conda-build runs it to build the Conda package.
ownconda develop runs it to install the package in editable mode.

ownconda develop also creates/updates a Conda environment for the current project and installs all requirements that it collects from the project’s recipe.

Projects also contain a .gitlab-ci.yml which defines the GitLab CI/CD pipeline. Most projects have at least a build, a test and an upload stage. The test stage is split into parallel steps for various test tools (e.g., pytest, pylint and bandit). Projects can optionally build documentation and upload it to our docs server. The ownconda tools provide helpers for all of these steps:

ownconda build builds the package.
ownconda test runs pytest.
ownconda lint runs pylint.
ownconda sec-check runs bandit.
ownconda upload uploads the package to the package index.
ownconda make-docs builds and uploads the documentation.

We also use our own Git flow:

Development happens in a develop branch. Builds from this branch are uploaded into a staging Conda channel.
Larger features can optionally branch of a feature branch. Their builds are not uploaded into a public Conda channel.
Stable develop states get merged into the master branch. Builds are uploaded into our stable Conda channel.
Since we continuously deploy packages, we don’t put a lot of effort into versioning. The package version consists of a major release which rarely changes and the number of commits since the last tagged major release. The GitLab pipeline ID is used as a build number:
- Version: $GIT_DESCRIBE_TAG.$GIT_DESCRIBE_NUMBER
- Build: py37_$CI_PIPELINE_ID
The required values are automatically exported by Conda and GitLab as environment variables.

Package and documentation hosting

Hosting a Conda repository is very easy. In fact, you can just run python -m http.server in your local Conda base directory if you previously built any packages. You can then use it like this: conda search --override-channels --channel=http://localhost:8000/conda-bld PKG.

A Conda repository consists of one or more channels. Each channel is a directory that contains a noarch directory and additional platform directories (like linux-64). You put your packages into these directories and run conda index channel/platform to create an index for each platform (you can omit the platform with newer versions of conda-build). The noarch directory must always exist, even if you put all your packages into the linux-64 directory.

The base URL for our Conda channels is https://forge.services.own/conda/channel. You can put a static index.html into each channel’s directory that parses the repo data and displays it nicely:

A JavaScript reads and renders the contents of a channel’s repodata.json.

The upload service (for packages created in GitLab pipelines) resides under https://forge.services.own/upload/<channel>. It is a simple web application that stores the uploaded file in channel/linux-64 and runs conda index. For packages uploaded to the stable channel, it also creates a hard link in a special archive channel.

Every week, we prune our channels with ownconda prune-index. In case that we accidentally prune too aggressively, we have the option to restore packages from the archive.

We also host our own Read the Docs like service. GitLab pipelines can upload Sphinx documentation to https://forge.services.own/docs via ownconda make-docs.

Python distribution

With Constructor, you can easily create your own self-extractable Python distribution. These distributions are similar to miniconda, but you can customize them to your needs.

A constructor file is a simple YAML file with some meta data (e.g., the distribution name and version) and the list of packages that should be included. You can also specify a post-install script.

The command constructor <distdir>/construct.yaml will then download all packages and put them into a self extracting Bash script. We upload the installer scripts onto our Conda index, too.

Instead of managing multiple construct.yaml files manually, we create them dynamically in a GitLab pipeline which makes building multiple similar distributions (e.g., for different Python versions) a bit easier.

Deployment

We are currently on the road from copy-stuff-with-fabric-to-vms to docker-kubernetes-yay-land. I am not going to go too much into detail here—this topic is not directly related to packaging and worth its own article.

Most of our deployments are now Ansible based. Projects contain an ansible directory with the required playbooks and other files. Shared roles are managed in a separate ownsible project. The ansible deployments are usually part of the GitLab CI/CD pipeline. Some are run automatically, some need to be triggered manually.

Some newer projects are already using Docker based deployments. Docker images are built as part of the pipeline and uploaded into our Docker registry from which they are then pulled for deployments.

Dependency management

It is very helpful if you can build a dependency graph of all your packages.

Not only can it be used to build all packages in the correct order (as we will shortly see), but visualizing your dependencies may also help you to improve your architecture, detect circular dependencies or unused packages.

The command ownconda dep-graph builds such a dependency graph from the packages that you pass to it. It can either output a sorted list of packages or a DOT graph. Since the resulting graph can become quite large, there are several ways to filter packages. For example, you can only show a package’s dependencies or why the package is needed.

The following figure shows the dependency graph for our python recipe. It was created with the command ownconda dep-graph external-recipes/ --implicit --requirements python --out=dot > deps_python.dot:

Dependency graph for Python

These graphs can become quite unclear relatively fast, though. This is the full dependency graph for the ownconda tools:

Dependency graph for the ownconda tools

I do not want to know how this would have looked if these were all JavaScript packages …

Making it work

Now that you know the theory of how everything should work, we can start to bootstrap our packaging infrastructure.

Some of the required steps are a bit laborious and you may need the assistance of your IT department in order to set up the domains and GitLab. Other steps can be automated and should be relatively painless, though:

Set up GitLab and a Conda repo server

Install GitLab. I’ll assume that it will be available under https://git.services.own.
Setup the forge server. I’ll assume that it will be available under https://forge.services.own:
- In your www root, create a conda folder which will contain the channels and their packages.
- Create the upload service that copies files sent to /upload/channel into www-root/conda/channel/linux-64 and calls conda index.
- Setup a Docker registry on the server.

Bootstrap Python, Pip and Conda

Clone all repositories that you need for the bootstrapping process:

$ mkdir -p ~/Projects/ownconda
$ cd ~/Projects/ownconda
$ for r in external-recipes ownconda-tools ownconda-dist; do \
>     git clone git@gitlab.com:ownconda/$r.git \
> done

Build all packages needed to create your Conda distribution. The ownconda tools provide a script that uses a Docker container to build all packages and upload them into the stable channel:
```
$ ownconda-tools/contrib/bootstrap.sh
```
Note

The script might fail to build some packages. The most probable causes are HTTP timeouts or unavailable servers. Just re-run the script and hope for the best. If the issue persists, you might need to fix the corresponding Conda recipe, though (Sometimes, people re-upload a source archive and thereby change its SHA256 value).

Create the initial Conda distributions and upload them:

$ cd ownconda-dist
$ python gen_installer.py .. 3.7
$ python gen_installer.py .. 3.7 dev
$ cd -
$ curl -F "file=@ownconda-3.7.sh" https://forge.services.own/upload/stable
$ curl -F "file=@ownconda-3.7-dev.sh" https://forge.services.own/upload/stable
$
$ # Create symlinks for more convenience:
$ ssh forge.services.own
# cd www-root/conda/stable
# ln -s linux-64/ownconda-3.7.sh
# ln -s linux-64/ownconda-3.7.sh ownconda.sh
# ln -s linux-64/ownconda-3.7-dev.sh
# ln -s linux-64/ownconda-3.7-dev.sh ownconda-dev.sh

You can now download the installers from https://forge.services.own/conda/stable/ownconda[-dev][-3.7].sh

Setup your local ownconda environment. You can use the installer that you just built (or (re)download it from the forge if you want to test it):

$ bash ownconda-3.7.sh
$ # or:
$ cd ~/Downloads
$ wget https://forge.services.own/conda/stable/ownconda-dev.sh
$ bash ownconda-dev.sh
$
$ source ~/.bashrc   # or open a new terminal
$ conda info
$ ownconda --help

Build the docker images

Create a GitLab pipeline for the centos7-ownconda-runtime project. This will generate your runtime Docker image.
When the runtime image is available, create a GitLab pipeline for the centos7-ownconda-develop project. This will generate your development Docker image used in your projects’ pipelines.

Build all packages

Create a GitLab pipeline for the external-recipes project to build and upload the remaining 3rd party packages.
You can now build the packages for your internal projects. You must create the pipelines in dependency order so that the requirements for each project are built first. The ownconda tools help you with that:
```
$ mkdir gl-projects
$ cd gl-projects
$ ownconda gitlab update
$ ownconda dep-graph --no-third-party --out=project . > project.txt
$ for p in $(cat projects.txt); do \
>     ownconda gitlab -p $p run-py ../ownconda-tools/contrib/gl_run_pipeline.py \
> done
```
If a pipeline fails and the script aborts, just remove the successful projects from the projects.txt and re-run the for loop.

Congratulations, you are done! You have built all internal and external packages, you have created your own Conda distribution and you have all Docker images that you need for running and building your packages.

Outlook / Future work and unsolved problems

Managing your organization’s packaging infrastructure like this is a whole lot of work but it rewards you with a lot of independence, control and flexibility.

We have been continuously improving our process during the last years and still have a lot of ideas on our roadmap.

While, for example, GitLab has a very good authentication and authorization system, our Conda repository lacks all of this (apart from IP restrictions for uploading and downloading packages). We do not want users (or automated scripts) to enter credentials when they install or update packages, but we are not aware of a (working) password-less alternative. Combining Conda with Kerberos might work in theory, but in practice this is not yet possible. Currently, we are experimenting with HTTPS client certificates. This might work well enough but it also doesn’t seem to be the Holy Grail of Conda Authorization.

Another big issue is creating more reproducible builds and easier rollback mechanisms in case an update ships broken code. Currently, we are pinning the requirements’ versions during a pipelines test stage. We are also working towards dockerized Blue Green Deployments and are exploring tools for container orchestration (like Kubernetes). On the other hand, we are still delivering GUI applications to client workstations via Bash scripts … (this works quite well, though, and provides us with a good amount of control and flexibility).

We are also still having an eye on Pip. Conda has the biggest benefits when deploying packages to VMs and client workstations. The more we use docker, the smaller the benefit might become, and we might eventually switch back to Pip.

But for now, Conda serves us very well.

The macOS Dark Mode, your Terminal and Vim

2018-10-23T22:24:00+02:00

The new Dark Mode in macOS Mojave is a nice addition and is – especially in the night hours — more pleasing to your eyes than the light mode.

MacOS light mode with a light Terminal profile and a light Vim theme.

However, enabling Dark Mode will not change the Terminal profile, which is a little bit annoying – especially if your color theme has a light and a dark variant (like the infamous Solarized, Snow, One, or my own Rasta theme).

MacOS dark mode with a light Terminal profile and a light Vim theme.

If you change your Terminal profile to something dark, Vim still doesn’t look right because it uses its own mechanism for light/dark backgrounds (see :help 'background' for details) and doesn’t know about the changes you made to the Terminal profile.

MacOS dark mode with a dark Terminal profile and a light Vim theme.

If you execute :set background=dark in Vim (and if you color scheme supports it), Vim looks nice and dark now, too.

MacOS dark mode with a dark Terminal profile and a dark Vim theme.

However, on the next day, the fun begins again when you want to switch everything back to light mode …

Wouldn’t it be nice if this could all be accomplished with a single command?

There are tools, that help you with switching to/from macOS Dark Mode (e.g., NightOwl or Shifty), but they can’t change your Terminal profile or notify Vim.

As it turns out, it’s not too hard to implement a little program that does exactly this:

You can uses the defaults command to get the current macOS Dark Theme mode:
```
$ defaults read -g AppleInterfaceStyle
Dark
```

You can use AppleScript (oh, how I love this language …) to set Dark Mode and update the Terminal profile:

# Set Dark Mode
tell application "System Events"
    tell appearance preferences
        set dark mode to true  # Can be one of: true, false, not dark
    end tell
end tell

# Update default settings (for new windows/tabs)
tell application "Terminal"
    set default settings to settings set "Rasta"
end tell

# Update settings for exsting windows/tabs
tell application "Terminal"
    set current settings of tabs of windows to settings set "Rasta"  # Theme name
end tell

You can wrap both things with a Python script:

# toggle-macos-dark-mode.py
import subprocess


OSASCRIPT = """
tell application "System Events"
    tell appearance preferences
        set dark mode to {mode}
    end tell
end tell

tell application "Terminal"
    set default settings to settings set "{theme}"
end tell

tell application "Terminal"
    set current settings of tabs of windows to settings set "{theme}"
end tell
"""

TERMINAL_THEMES = {
    False: 'Rasta light',
    True: 'Rasta',
}


def is_dark_mode() -> bool:
    """Return the current Dark Mode status."""
    result = subprocess.run(
        ['defaults', 'read', '-g', 'AppleInterfaceStyle'],
        text=True,
        capture_output=True,
    )
    return result.returncode == 0 and result.stdout.strip() == 'Dark'


def set_interface_style(dark: bool):
    """Enable/disable dark mode."""
    mode = 'true' if dark else 'false'  # mode can be {true, false, not dark}
    script = OSASCRIPT.format(mode=mode, theme=TERMINAL_THEMES[dark])
    result = subprocess.run(
        ['osascript', '-e', script],
        text=True,
        capture_output=True,
    )
    assert result.returncode == 0, result


if __name__ == '__main__':
    set_interface_style(not is_dark_mode())

You can use the timer_start() function introduced in Vim 8 and neovim to regularly check for the current Dark Mode settings. Put this into your Vim config:

function! SetBackgroundMode(...)
    let s:new_bg = "light"
    if $TERM_PROGRAM ==? "Apple_Terminal"
        let s:mode = systemlist("defaults read -g AppleInterfaceStyle")[0]
        if s:mode ==? "dark"
            let s:new_bg = "dark"
        else
            let s:new_bg = "light"
        endif
    else
        " This is for Linux where I use an environment variable for this:
        if $VIM_BACKGROUND ==? "dark"
            let s:new_bg = "dark"
        else
            let s:new_bg = "light"
        endif
    endif
    if &background !=? s:new_bg
        let &background = s:new_bg
    endif
endfunction
call SetBackgroundMode()
call timer_start(3000, "SetBackgroundMode", {"repeat": -1})

You can create an Automator action that runs the Python script and that can be activated with a global shortcut. I use ⌥⌘D (you need to deactivate this shortcut for showing/hiding the Dock first). This is the AppleScript I used:
```
do shell script "/usr/local/bin/python3 ~/toggle-macos-dark-mode.py"
```

The drawback of this method is that the current application (at the time you press ⌥⌘D) is used as “source” of the action you get two dialogs asking you to give that app permissions to remote control the System Settings and Terminal.

A better solution would be if the authors of NightOwl and Shifty would integrated this into their tools. I’m gonna contact them and see what happens. :-)

Update:

MacVim got an OSAppearanceChanged event that is emitted every time MacVim changes its appearance.

Thanks to Frank for the heads up!

How to mask an image with a smooth circle in PyQt5

2018-05-04T13:37:00+02:00

I am currently working on a PyQt5 based user manager and want to display user images as circle.

User image can have any size and aspect ratio, so I also need to crop them before applying the mask:

You can easily set a QPixmap to a QLabel and show it in your widget. However, it took me hours to find out how to mask an image with a smooth, anti-aliased circle.

If you search for this problem, you’ll find some (relatively old) StackOverflow questions and topics in Qt5 forums, but none of the proposed solutions worked for me.

The following snipped does:

crop the original image to a square
mask it with a smooth, anti-aliased circle
rescale it to the required size
take care of Retina / high DPI displays

from PyQt5.QtCore import Qt, QRect
from PyQt5.QtGui import QBrush, QImage, QPainter, QPixmap, QWindow
from PyQt5.QtWidgets import QLabel, QVBoxLayout, QWidget


def mask_image(imgdata, imgtype='jpg', size=64):
    """Return a ``QPixmap`` from *imgdata* masked with a smooth circle.

    *imgdata* are the raw image bytes, *imgtype* denotes the image type.

    The returned image will have a size of *size* × *size* pixels.

    """
    # Load image and convert to 32-bit ARGB (adds an alpha channel):
    image = QImage.fromData(imgdata, imgtype)
    image.convertToFormat(QImage.Format_ARGB32)

    # Crop image to a square:
    imgsize = min(image.width(), image.height())
    rect = QRect(
        (image.width() - imgsize) / 2,
        (image.height() - imgsize) / 2,
        imgsize,
        imgsize,
    )
    image = image.copy(rect)

    # Create the output image with the same dimensions and an alpha channel
    # and make it completely transparent:
    out_img = QImage(imgsize, imgsize, QImage.Format_ARGB32)
    out_img.fill(Qt.transparent)

    # Create a texture brush and paint a circle with the original image onto
    # the output image:
    brush = QBrush(image)        # Create texture brush
    painter = QPainter(out_img)  # Paint the output image
    painter.setBrush(brush)      # Use the image texture brush
    painter.setPen(Qt.NoPen)     # Don't draw an outline
    painter.setRenderHint(QPainter.Antialiasing, True)  # Use AA
    painter.drawEllipse(0, 0, imgsize, imgsize)  # Actually draw the circle
    painter.end()                # We are done (segfault if you forget this)

    # Convert the image to a pixmap and rescale it.  Take pixel ratio into
    # account to get a sharp image on retina displays:
    pr = QWindow().devicePixelRatio()
    pm = QPixmap.fromImage(out_img)
    pm.setDevicePixelRatio(pr)
    size *= pr
    pm = pm.scaled(size, size, Qt.KeepAspectRatio, Qt.SmoothTransformation)

    return pm


class Window(QWidget):
    """Simple window that shows our masked image and text label."""
    def __init__(self, imgpath):
        super().__init__()

        imgdata = open(imgpath, 'rb').read()
        pixmap = mask_image(imgdata)
        ilabel = QLabel()
        ilabel.setPixmap(pixmap)
        tlabel = QLabel('Hello, world!')

        layout = QVBoxLayout()
        layout.addWidget(ilabel, 0, Qt.AlignCenter)
        layout.addWidget(tlabel, 0, Qt.AlignCenter)
        self.setLayout(layout)


if __name__ == '__main__':
    import sys
    from PyQt5.QtWidgets import QApplication

    app = QApplication(sys.argv)
    w = Window(imgpath=sys.argv[1])
    w.show()
    sys.exit(app.exec_())

And this is how the result looks like:

I also toyed around with QPainter.setCompositionMode() and QPixmap.setMask() but both approaches did not lead to the desired results.

If you have found a better solution for this problem, please let me know via Mastodon or Twitter.

Assertions and Exceptions

2018-04-16T09:26:00+02:00

I recently had a discussion about the difference of Python’s assert keyword vs. raising “normal” exceptions. For quite a long time, I was uncertain when to use assert, too. In this article, I’ll present some examples and rules of thumb for when using assert or raising exceptions is – in my opinion – more appropriate.

When do I use `assert`?

You use assert to check and document invariants of your code. Invariants are conditions in the middle of your code that should always be true (For example: “At this point in time x should always be a number between 0 and 10”).

Assertions can help others (including future-you) to understand your code and they should make your program crash if the assertion becomes false (because of a programming error).

Here are some examples for when assert is useful:

# Bad:

def func(a, b):
    x = [a, b]
    ...  # Some operations on "x"
    # x should still contain 2 elements:
    return *x


# Good:

def func(a, b):
    x = [a, b]
    ...  # Some operations on "x"
    assert len(x) == 2, f'len(x) == {len(x)}'
    return *x

# We have widget like this:
#
#   Step size: ( ) auto   (•) manual: [___300] seconds

# Bad:

if self.button_auto.isChecked():
    return None
else:
    # The user provided a manual step size:
    return self.step_size_input.value()


# Good:

if self.button_auto.isChecked():
    return None
else:
    assert self.button_manual.isChecked()
    return self.step_size_input.value()

# Bad:

def fill_list(self):
    # self.dest should be empty, but self.source not:
    while self.source_not_empty():
        self.dest.append(self.take_from_source())


# Good:

def fill_list(self):
    assert len(self.dest) == 0
    assert len(self.source) > 0
    while self.source_not_empty():
        self.dest.append(self.take_from_source())
    assert len(self.dest) > 0

Validation of input parameters

Sometimes, assert is used to validate input parameters for functions. You should avoid this because assert statements are not executed when Python is invoked with -O[O]. In this case, you would suddenly loose all input validation.

AssertionErrors also carry less semantics than “normal” exceptions – like TypeError when the user passed a string and not a number, or ValueError when the user passed a negative number when only positive numbers are allowed.

# Bad:

def sum(a, b):
    assert isinstance(a, int), 'a must be an int'
    assert a >= 0, 'a must be >= 0'
    ...

# Good:

def sum(a, b):
    if not isinstance(a, int):
        raise TypeError(f'a must be an int but is a {type(a)}')
    if a < 0:
        raise ValueError(f'a must be >= 0 but is {a}')
    ...

There are, however, exceptions possible for this rule. Imagine the following snippet where we want to add/delete a user from a group on an LDAP server:

def add_to_group(conn, user, group):
    _modify_group(conn, user, group, ldap3.MODIFY_ADD)

def delete_from_group(conn, user, group)
    _modify_group(conn, user, group, ldap3.MODIFY_DELETE)

def _modify_group(conn, user, group, op):
    assert op in {ldap3.MODIFY_ADD, ldap3.MODIFY_DELETE}
    conn.modify(group.dn, {'uniqueMember': [(op, [user.username])]})

In this case, it’s appropriate to use assert because _modify_group() should never be called by our users but only by add_to_group() and delete_from_group().

What kind of exceptions should I raise?

Within you own code, the built-in exception types – especially ValueError, AttributeError, TypeError and RuntimeError – are usually sufficient.

Libraries often define their own exception hierarchy with their own base class. For example, Requests has RequestError from which all other exception types inherit.

This makes it a lot easier for users to catch your library’s exceptions. If you raise all kinds of exceptions (and then you don’t document that, too), your users will end up using a broad/bare excepts which is bad.

except [[Base]Exception] is bad because it results in all kinds of nasty side effects. For example, it would also catch (and possibly ignore) KeyboardInterrupt (Ctrl+C) or AsstionError.

And that’s why you should also never raise [Base]Exception.

TLDR

Use assert to detect programming errors and conditions that should never occur and which should crash your program immediately, e.g., invariants of an algorithm.

Warning: Assertion checks are stripped and not executed if Python is invoked with the -O or --OO option!
Raise an exception for errors that are caused by invalid user input or other problems with the environment (e.g., network errors or unreadable files).
Never raise or catch [Base]Exception.

Getting started with devpi

2017-11-09T21:33:00+01:00

Recently, I wanted to (re)evaluate devpi for use in our company. I have already worked with it some years ago but–by now–forgot what exactly it can do and how to set it up.

However, the marketing on the landing page of the docs and on GitHub was not very convincing and I nearly ended up working with another product.

A short Twitter discussion later, I decided to give devpi a try and write down my findings. Maybe they can contribute to improving devpi’s docs and demonstrate how easy it is to use.

This is what I was trying to achieve with devpi:

Mirror and cache PyPI
Extend PyPI with an index for internal stable packages
Extend the stable index with a staging index for experimental new features
No free registration for users
Only a single, authorized user (our GitLab CI)
Use standard tools (pip, twine, …) as much as possible

Installation

Devpi consists of a server, a command line client and a web front-end. The meta package devpi installs them all:

$ mkvirtualenv devpi
$ pip install devpi
...
Installing collected packages: ...
Successfully installed ...

Setting up the server

Devpi uses SQLite to store its data, so we don’t need to setup a database (you can use PostgreSQL though, if you want).

When you start the devpi-server for the first time, you need to pass the --init option. You an also specify where it should store its data (the default is ~/.devpi):

$ devpi-server --serverdir=/tmp/devpi --init
...

If you don’t use the standard location, you must set the --serverdir option every time you start the server.

User management

We can now set a password for the root user and allow only root to create new users:

$ devpi-server --serverdir=/tmp/devpi --passwd root
enter password for root:
repeat password for root:

$ devpi-server --serverdir=/tmp/devpi --restrict-modify=root --start
...
starting background devpi-server at http://localhost:3141
...

Like the --serverdir option, you must always pass --restrict-modify=root when you start the server.

Once the server is running, we can use the devpi client devpi to create an additional user named packages. Prior to that, tell the devpi client on which server we want to operate with the following commands:

$ devpi use http://localhost:3141
...

$ devpi login root
password for user root:
logged in 'root', credentials valid for 10.00 hours

$ devpi user -c packages email=packaging@company.com password=packages
user created: packages

$ devpi user -l
packages
root

Package indexes

By default, devpi creates an index called root/pypi. It serves as a proxy and cache for PyPI and you can’t upload your own packages to it.

However, devpi supports index inheritance: We can create our stable index in the packages namespace and set root/pypi as base. If we query packages/stable, devpi first searches this index and than falls back to root/pypi if it can’t find the package on the first index.

$ devpi index -c packages/stable bases=root/pypi volatile=False
http://localhost:3141/packages/stable:
  type=stage
  bases=root/pypi
  volatile=False
  acl_upload=packages
  mirror_whitelist=
  pypi_whitelist=

Similarly, our staging index can inherit stable. Devpi will than search packages/staging, packages/stable and finally root/pypi for packages:

$ devpi index -c company/staging bases=company/stable volatile=True
http://localhost:3141/company/staging:
  type=stage
  bases=company/stable
  volatile=True
  acl_upload=company
  mirror_whitelist=
  pypi_whitelist=

The volatile=True option lets us perform destructive actions on the index (like overriding or deleting packages).

Install packages

Let’s use our devpi to load a public package from PyPI:

$ pip install -i http://localhost:3141/company/stable click
Collecting click
  Downloading http://localhost:3141/root/pypi/+f/5e7/a4e296b3212da/click-6.7-py2.py3-none-any.whl (71kB)
Installing collected packages: click
Successfully installed click-6.7

We can also configure pip to use our devpi as default index:

[global]
index-url = http://localhost:3141/company/stable

Upload packages

You can build and upload packages with your usual workflow. Just add devpi to your ~/.pypirc (Do not store passwords in there!):

[distutils]
index-servers =
    devpi-stable
    devpi-staging

[devpi-stable]
repository = http://localhost:3141/packages/stable/
username = packages

[devpi-staging]
repository = http://localhost:3141/packages/staging/
username = packages

Now we can build some dists:

$ cd ~/Projects/simpy
$ python setup.py sdist bdist_wheel
...

And upload them:

$ pip install twine
...
Installing collected packages: ...
Successfully installed ...

$ twine upload -r devpi-stable dist/*
Uploading distributions to http://localhost:3141/packages/stable/
Enter your password:
Uploading simpy-3.0.10-py2.py3-none-any.whl
Uploading simpy-3.0.10.tar.gz

Congratulation!

You can now install your own packages from your own packages index:

$ pip install simpy
Collecting simpy
  Downloading simpy-3.0.10-py2.py3-none-any.whl
Installing collected packages: simpy
Successfully installed simpy-3.0.10

Devpi’s docs may appear a bit confusing and look a little demure, but devpi itself is actually really easy to setup and use – and powerful at the same time!

Advanced asyncio testing

2016-03-10T16:22:00+01:00

In my last article, I showed how pytest’s fixture system and plug-in infrastructure can help you with writing cleaner and better tests. Fixtures allow you to create a clean event loop instance for every test case. The plug-in system allows you to write test functions that are actually asyncio coroutines. While I was working on that articel, Tin Tvrtkovic created the plug-in pytest-asyncio.

In short, it lets you do this:

import asyncio
import time

import pytest

@pytest.mark.asyncio
async def test_coro(event_loop):
    before = time.monotonic()
    await asyncio.sleep(0.1, loop=event_loop)
    after = time.monotonic()
    assert after - before >= 0.1

Instead of this:

import asyncio
import time

def test_coro():
    loop = asyncio.new_event_loop()
    try:
        asyncio.set_event_loop(loop)

        before = time.monotonic()
        loop.run_until_complete(asyncio.sleep(0.1, loop=loop))
        after = time.monotonic()
        assert after - before >= 0.1
    finally:
        loop.close()

So using pytest-asyncio clearly improves your test (and there is even more, what this plug-in does!).

While I have been working on aiomas, some additional requirements came up that were not so easily covered. What aiomas basically does is adding three layers of abstraction around the asyncio transports:

The channel layer lets you send JSON or MsgPack encoded messages in a request-reply manner. This layer uses a custom protocol that works with different kinds of transports: TCP sockets, Unix domain sockets and custom transport called local queue.
The RPC layer creates a remote-procedure-call system on top of the channel layer.
The agent layer (for multi-agent systems) hides even more of the networking-related stuff and lets you basically write classes that call methods of other classes over a network connection.

Here is a simple example of how the channel layer works:

import aiomas


async def handle_client(channel):
    """Handle a client connection."""
    req = await channel.recv()
    print(req.content)
    await req.reply('cya')
    await channel.close()


async def client():
    """Client coroutine: Send a greeting to the server and wait for a
    reply."""
    channel = await aiomas.channel.open_connection(('localhost', 5555))
    rep = await channel.send('ohai')
    print(rep)
    await channel.close()


server = aiomas.run(aiomas.channel.start_server(
    ('localhost', 5555), handle_client))
aiomas.run(client())

server.close()
aiomas.run(server.wait_closed())

Requirements for our tests

So with this in mind, I had the following requirements for my tests:

I need a clean event loop instance for every test.

This can be solved with the event_loop fixture provided by pytest-asyncio.
Every test should be run with every transport available (TCP socket, Unix domain socket, …).

This could in theory be solved with the pytest.mark.parametrize() decorator (but not in my case as we will see later).
Every test needs a client coroutine. Ideally, this would be the test itself.

pytest-asyncio’s pytest.mark.asyncio decorator solves this.
Every test needs a server with a custom callback for client connections. Servers must be cleanly shut down no matter what the outcome of the test is.

It would seem that a fixture would do the job, but every server needs a test specific callback for handling client connections. This makes it a lot harder.
I don’t want any “address already in use” errors if one test fails badly.

pytest-asyncio’s unused_tcp_port fixture comes to help.
I don’t want to use loop.run_until_complete() all the time.

Again, the pytest.mark.asyncio decorator solves this.

To wrap up what remains to be solved: Every test needs at least two fixtures (one for the event loop, one for the address type), but I want to combine them as a single fixture. Creating a fixture for setting up a server would also be nice, but how can we do this?

Our first approach

The first thing we can do is to wrap the loop and the address type in a fixture. We’ll call it ctx (short for test context). With fixture parameters, it is also easy to create one fixture instance for every address type:

import tempfile
import py
import pytest


class Context:
    def __init__(self, loop, addr):
        self.loop = loop
        self.addr = addr


@pytest.fixture(params=['tcp', 'unix'])
def ctx(request, event_loop, unused_tcp_port, short_tmpdir):
    """Generate tests with TCP sockets and Unix domain sockets."""
    addr_type = request.param
    if addr_type == 'tcp':
        addr = ('127.0.0.1', unused_tcp_port)
    elif addr_type == 'unix':
        addr = short_tmpdir.join('sock').strpath
    else:
        raise RuntimeError('Unknown addr type: %s' % addr_type)

    ctx = Context(event_loop, addr)
    return ctx


@pytest.yield_fixture()
def short_tmpdir():
    """Generate a short temp. dir for Unix domain sockets.  The paths
    provided by ptest's tmpdir fixture are too long on some platforms."""
    with tempfile.TemporaryDirectory() as tdir:
        yield py.path.local(tdir)

This lets us write our tests like this:

import aiomas

@pytest.mark.asyncio
async def test_channel(ctx):
    results = []

    async def handle_client(channel):
        req = await channel.recv()
        results.append(req.content)
        await req.reply('cya')
        await channel.close()


    server = await aiomas.channel.start_server(ctx.addr, handle_client)
    try:
        channel = await aiomas.channel.open_connection(ctx.addr)
        rep = await channel.send('ohai')
        results.append(rep)
        await channel.close()

    finally:
        server.close()
        await server.wait_closed()

    assert results == ['ohai', 'cya']

This works already very nicely and every test using the ctx fixture is run once for every address type.

However, two problems remain:

Our ctx fixture always requires an unused TCP port and a temporary directory – although we only need one of both in each case.
Setting up the server (and closing it) also involves some code which will be the same for every test and should thus be moved into a fixture. However, a server fixture won’t work directly, because every server needs a test specific callback as you can see in the line where we create the server (server = await ...). But without a server fixture, we can’t have a tear-down method for it …

Let’s see how we can tackle these issues.

Approach number two

The first problem can be solved by using the getfuncargvalue() method of the request object that our fixture receives. Using this method, we can manually call a fixture function:

@pytest.fixture(params=['tcp', 'unix'])
def ctx(request, event_loop):
    """Generate tests with TCP sockets and Unix domain sockets."""
    addr_type = request.param
    if addr_type == 'tcp':
        port = request.getfuncargvalue('unused_tcp_port')
        addr = ('127.0.0.1', port)
    elif addr_type == 'unix':
        tmpdir = request.getfuncargvalue('short_tmpdir')
        addr = tmpdir.join('sock').strpath
    else:
        raise RuntimeError('Unknown addr type: %s' % addr_type)

    ctx = Context(event_loop, addr)
    return ctx

To help with issue number two, we can extend our Context class that is passed into every test. We add a method Context.start_server(client_handler) that we can call from within our tests. We also add a finalize/teardown part to our ctx fixture that will close the server once we are done. And while we are at it, we’ll also create some more shortcut functions:

import asyncio
import tempfile
import py
import pytest


class Context:
    def __init__(self, loop, addr):
        self.loop = loop
        self.addr = addr
        self.server = None

    async def connect(self, **kwargs):
        """Create and return a connection to "self.addr"."""
        return (await aiomas.channel.open_connection(
            self.addr, loop=self.loop, **kwargs))

    async def start_server(self, handle_client, **kwargs):
        """Start a server with the callback *handle_client* listening on
        "self.addr"."""
        self.server = await aiomas.channel.start_server(
            self.addr, handle_client, loop=self.loop, **kwargs)

    async def start_server_and_connect(self, handle_client,
                                       server_kwargs=None,
                                       client_kwargs=None):
        """Shortcut for::

            await ctx.start_server(...)
            channel = await ctx.connect()"

        """
        if server_kwargs is None:
            server_kwargs = {}

        if client_kwargs is None:
            client_kwargs = {}

        await self.start_server(handle_client, **server_kwargs)
        return (await self.connect(**client_kwargs))

    async def close_server(self):
        """Close the server."""
        if self.server is not None:
            server, self.server = self.server, None
            server.close()
            await server.wait_closed()


@pytest.yield_fixture(params=['tcp', 'unix'])
def ctx(request, event_loop):
    """Generate tests with TCP sockets and Unix domain sockets."""
    addr_type = request.param
    if addr_type == 'tcp':
        port = request.getfuncargvalue('unused_tcp_port')
        addr = ('127.0.0.1', port)
    elif addr_type == 'unix':
        tmpdir = request.getfuncargvalue('short_tmpdir')
        addr = tmpdir.join('sock').strpath
    else:
        raise RuntimeError('Unknown addr type: %s' % addr_type)

    ctx = Context(event_loop, addr)

    yield ctx

    # Shutdown the server and wait for all pending tasks to finish:
    aiomas.run(ctx.close_server())
    aiomas.run(asyncio.gather(*asyncio.Task.all_tasks(event_loop),
                              return_exceptions=True))

With this extra functionality, our test case becomes a lot shorter, easier to read, and more reliable:

import aiomas

@pytest.mark.asyncio
async def test_channel(ctx):
    results = []

    async def handle_client(channel):
        req = await channel.recv()
        results.append(req.content)
        await req.reply('cya')
        await channel.close()


    channel = await ctx.start_server_and_connect(handle_client)
    rep = await channel.send('ohai')
    results.append(rep)
    await channel.close()

    assert results == ['ohai', 'cya']

The ctx fixture (and the associated Context class) is indeed not the shortest fixture I ever wrote, but it helped me to remove approx. 200 lines of boilerplate code from my tests (apart from making them more readable and maintainable).

Testing (asyncio) coroutines with pytest

2015-04-22T20:34:00+02:00

Pytest is an awesome testing package for Python and since long one of my favorite Python packages in general. It makes writing test really easy and its reporting capabilities for test failures are extremely helpful.

However, it currently (as of version 2.7) doesn’t help you very much with testing (asyncio) coroutines. So a naïve approach for testing coroutines would be:

# tests/test_coros.py

import asyncio

def test_coro():
    loop = asyncio.get_event_loop()

    @asyncio.coroutine
    def do_test():
        yield from asyncio.sleep(0.1, loop=loop)
        assert 0  # onoes!

    loop.run_until_complete(do_test())

This approach has several problems and a lot of overhead. The only interesting lines are the ones containing the yield from and assert statements.

It would be better if every test case had its own event loop instance that gets correctly closed no matter if the test passes or fails.

We also cannot simply yield within our test case because pytest would then think that our test case yields new test cases which is not the case here. So we have to create a separate coroutine which contains our actual test and fire up the even loop in order to execute it.

Our tests would look a lot cleaner and behave better if we could instead do something like this:

# tests/test_coros.py

@asyncio.coroutine
def test_coro(loop):
    yield from asyncio.sleep(0.1, loop=loop)
    assert 0

It turns out that thanks to pytests flexible plug-in system, it’s possible to implement the desired behavior. Although, most of the required hooks are not documented very well or at all which also makes it relatively hard to find out which hooks you have to implement and how.

We create a local per-directory plug-in since this is a little bit easier than creating a “real”, external plug-in. Pytest looks in every test-directory for a file called conftest.py and applies the fixtures and hooks implemented there to all tests within that directory.

So lets start by writing a fixture that creates a new event loop instance for each test case and properly close it when the test is done:

# tests/conftest.py

import asyncio
import pytest


@pytest.yield_fixture
def loop():
    # Set-up
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    yield loop

    # Clean-up
    loop.close()


# tests/test_coros.py

def test_coro(loop):
    @asyncio.coroutine
    def do_test():
        yield from asyncio.sleep(0.1, loop=loop)
        assert 0  # onoes!

    loop.run_until_complete(do_test())

Before each test, pytest executes the loop fixture until the first yield statement. What gets yielded is then passed to the loop argument of our test case. When the test ends (successfully or not), pytest finishes the execution of the loop fixture and thus closes the loop properly. In the same way, you could write a fixture that creates a socket and closes it after each test (your socket fixture can depend on the loop fixture in the same way as our test does. Nice, isn’t it?).

But we are still not done yet. Let’s teach pytest how to execute our test coroutines. Therefore, we need to change how asyncio coroutines are collected (they should be collected like normal test functions, not like test generators) and how they get executed (via loop.run_until_complete()):

# tests/conftest.py

def pytest_pycollect_makeitem(collector, name, obj):
    """Collect asyncio coroutines as normal functions, not as generators."""
    if collector.funcnamefilter(name) and asyncio.iscoroutinefunction(obj):
        # We return a list of test function objects.  Depending on the
        # fixtures, one test function can result in multiple test items
        # (i.e., when it is decorated with "pytest.mark.parametrize()".
        return list(collector._genfunctions(name, obj))
    # else:
    #     We return None and pytest's default behavior gets applied to "obj"


def pytest_pyfunc_call(pyfuncitem):
    """If ``pyfuncitem.obj`` is an asyncio coroutinefunction, execute it via
    the event loop instead of calling it directly."""
    testfunction = pyfuncitem.obj

    if not asyncio.iscoroutinefunction(testfunction):
        # Return None if its not a coroutine.  Pytest will handle the
        # test item in the normal way.
        return

    # Extract required arguments from all available fixtures:
    funcargs = pyfuncitem.funcargs  # dict with all fixtures
    argnames = pyfuncitem._fixtureinfo.argnames  # Args for this test
    testargs = {arg: funcargs[arg] for arg in argnames}

    # Create the generator object for this test (test will not yet execute!)
    coro = testfunction(**testargs)

    # Run the coro in the event loop and execute the test
    loop = testargs['loop'] if loop in testargs else asyncio.get_event_loop()
    loop.run_until_complete(coro)

    return True  # Signal pytest that we executed the test

This plug-in should work with pytest 2.4 or newer. I tested it with versions 2.6 and 2.7.

It seems like the pytest devs are currently trying to clean-up and refactor pytest’s plug-in system. Maybe, a future release will even contain similar functionality to our plug-in. It would be really nice if pytest supported testing coroutines out-of-the-box as they are becoming increasingly popular amongst various frameworks.

Update: There’s also pytest-asyncio . It was created after I posted the solution described above to Stack Overflow.

Thanks to ronny and hpk for proof-reading this article.

aiomas – A library for multi-agent systems and RPC based on asyncio

2015-02-13T10:40:00+01:00

aiomas is a new, asyncio based library that helps you with creating distributed multi-agent systems. It allows different clocks (real-time, coupled to simulations, …) and different codecs (MsgPack, JSON) to be used. You can also just use the lower level abstractions for RPC or message based request-reply channels.

I started the development in December 2014 and OFFIS, the institute I’m working at, now kindly allowed me to release it as OSS (licensed under the MIT license).

What? But why?

Multi-agent systems are a thing in many scientific institutes, including OFFIS. In particular, my colleagues and I want to create a multi-agent system for managing dynamic virtual power plants in our current project Dynamic VPP (German link).

A very common tool (at least in science) for writing multi-agent systems is JADE. However, most people I know don’t really like working with JADE (yes, that’s an euphemism).

I also didn’t find a modern, well documented and easy-to-use Python library for the job. So I asked my self: “How hard would it be to create one on my own?”

Being one of the maintainers of the discrete-event simulation (DES) library SimPy, I already have some experience with event-based and asynchronous programming. Multi-agent systems are relatively similar to DES, so I thought I should give it a try.

I also wanted to get to know asyncio. I’ve never worked with it before but its concepts looked very similar to the ones we’ve been using in SimPy for years, so building a library for multi-agent systems on top of asyncio was a perfect project to learn something new.

How do I use it?

The basic idea behind aiomas is that agents should be plain Python classes. Methods that perform asynchronous operations (like sending messages to other agents) should be coroutines. Agents should be able to call methods that other agents expose. It should also be easy to distribute agents over multiple CPU cores or machines. But you, as the user, shouldn’t bothered with setting up sockets or implementing transports, protocols and other low-level stuff.

aiomas is still in a relatively early stage of development. What’s already there seems to work quite nicely, though.

Here is a simple example how it currently looks like:

>>> import asyncio
>>> import aiomas
>>>
>>> class TestAgent(aiomas.Agent):
...     def __init__(self, container, name):
...         super().__init__(container, name)
...         print('Ohai, I am %s' % name)
...
...     @asyncio.coroutine
...     def run(self, addr):
...         """The agent's "main" function."""
...         remote_agent = yield from self.container.connect(addr)
...         ret = yield from remote_agent.service(42)
...         print('%s got %s from %s' % (self.name, ret, addr))
...
...     @aiomas.expose
...     def service(self, value):
...         """Exposed function that can be called by remote agents."""
...         return value
>>>
>>> # A Container is the home for a number of agents.
>>> c = aiomas.Container(('localhost', 5555))
>>> agents = [c.spawn(TestAgent) for i in range(2)]
Ohai, I am agent://localhost:5555/0
Ohai, I am agent://localhost:5555/1
>>>
>>> # Run agent 0 and let it call a method from agent 1
>>> loop = asyncio.get_event_loop()
>>> loop.run_until_complete(agents[0].run('agent://localhost:5555/1'))
agent://localhost:5555/0 got 42 from agent://localhost:5555/1
>>>
>>> c.shutdown()

What’s the list of features?

aiomas just puts three layers of abstraction around raw TCP / unix domain sockets provided by asyncio:

Request-reply channel:

The channel layer is the basis for the rpc layer. It sends JSON or MsgPack encoded byte strings over TCP or unix domain sockets. It also maps replies (of success or failure) to their corresponding request.

RPC:

The rpc layer implements remote procedure calls which let you call methods on remote objects nearly as if they were normal objects:

Instead of ret = obj.meth(arg) you write ret = yield from obj.meth(arg).

Agents and agent containers:

The top-layer provides a simple base class for your own agents. All agents live in a container.

Containers take care of creating agents and performing the communication between them.

The container provides a clock for the agents. This clock can either be synchronized with the real (wall-clock) time or be set by an external process (e.g., other simulators).

Although you usually want to use the agent layer, it is perfectly okay to only use the rpc or channel layer.

What’s comming next?

Some ideas for future releases:

SSL/TLS support for TCP sockets
Optional automatic re-connect after connection loss
Helper for binding a socket to a random free port
Implement a queue based transport that agents within a single container can use instead of a TCP connection.

Installation

aiomas requires Python >= 3.4, msgpack-python and arrow (it may also run on Python 3.3 with the asyncio package, but this is untested).

Install aiomas via pip by running:

$ pip install aiomas

The development happens on bitbucket. There’s no mailing list or IRC channel yet, so just create an issue for your feedback, send me an email, or leave me a tweet.

Master(ing) Passwords

2015-02-01T13:37:00+01:00

Good passwords should be only known to you, they should be hard to guess, you should have different ones for each service and you often need to change them. Meeting all these requirements is hard and people often fail.

Some people claim that biometrics (like finger prints, facial recognition or iris scans) can help you out of this misery, but they are wrong. You can’t keep your face (and even your finger prints) private. Worse, you can’t change them if someone starts to misuse them. So they can, at best, be used as a kind of user name.

You can use password safes to store complicate passwords for all the web service you use. But your app may use a proprietary container format. What do you do if the app becomes obsolete and all of your passwords are stored in a format no other software can read? And can you really be sure that no one can read your passwords that you uploaded into the cloud?

A relatively new approach to solve this problem is the Master Password algorithm by Maarten Billemont. Its basic idea is strikingly simple: You have one (complicated) password that you have to remember. Based on that password, the algorithm generates site or service specific passwords of any complexity – ranging from four digit PINs to 20 characters long all-random passwords. Given the same input, the algorithm always generates the same passwords. Since there’s only one secret involved in the whole process (your master password that you keep in your head), no passwords need to be stored anywhere and yet you can use Master Password from all your devices if you have an appropriate app available.

I read about the Master Password algorithm in August 2014 in the German IT magazine c’t and got curious how hard it would be to implement the algorithm in Python (3). It turned to be quite straightforward. After toying around with it for a while, I have now a relatively stable and useful master password library and, on top of it, a command line interface.

Available implementations

Before I go into detail about it, I want to list all available implementations I’m aware of (please contact me if I forgot one):

Official implementations (the source code for all of them can be found on github):
- OS X GUI: It is available via the Mac App Store.
- iOS App: It is available via the iTunes Store.
- Java GUI: Supports Mac, Linux, Unix, and Windows and can be downloaded from the official website.
- Native C CLI: Supports Mac, Linux, and Posix and can be downloaded from the official website.
- Android App (beta): Available via the official website.
- JavaScript web app (beta): Available via the official website.

Alternative implementations:
- Master Password for Android: An alternative Android client with a nicer GUI (in Lollipop’s Material Design). Available as free and pro version in Google’s Play Store.
- qMasterPassword: A QT based GUI written in C++. It should be platform independent but unfortunately doesn’t provide any binaries. You can grab the source from github and compile it on your own.
- Chrome Extension: A Chrome extension available in the Chrome Web Store. The source can be found on github.
- Python CLI and library: My own implementation in Python. It runs on OS X, Linux and Windows. You can get it via the PyPI. The sources are on bitbucket.

As you can see, you can use Master Password on most of the common platforms and devices.

mpw 0.3

After two prototypes, my mpw package has now reached a certain level of maturity and provides some comfort. It’s still labeled as alpha, though, because some features I’d like to implement are still missing.

For your convenience, mpw remembers the sites and users that you created, so the first thing you have to do is creating a user (you can have multiple) and some sites for them:

$ mpw adduser Alice
Enter master password:
Confirm master password:
Added user "Alice".
$ mpw addsite test-site
Enter master password for "Alice":
Added site "test-site" for user "Alice".

mpw stores an scrypt hash of your master key. It will then tell you when you make typos while entering your master password. By default, newly created sites will use the long password type. Run mpw addsite --help to get more information on that. The sites and their settings are stored encrypted on your disk (using cryptography Fernet). Note, that these steps only simplify the later usage and are not required by the algorithm.

To actually generate a password for Alice’ test-site account, run:

$ mpw get test-site
Enter master password for "Alice":
Password for "test-site" for user "Alice" was copied to the clipboard.

As the output suggest, the command copies the password into your clipboard. You can then simply paste it into a login form.

mpw can also print the generated password to stdout. This way you can pipe it directly to other commands like sudo:

$ mpw get -e server-root | sudo -S vim /etc/crontab

The command mpw --help prints a list of all commands that are currently supported.

Here is a list of all available features:

Supports multiple versions of the algorithm. If the algorithm evolves, you can still generated password for older sites using an older version of the algorithm. Currently, version 1 and 2 are implemented.
Generated passwords are copied into your clipboard by default. You can also print them to stdout and pipe them into other commands.
Lets you add, delete and list users. You can set a user as default user and you can change their master password (which will invalidate, by the very nature of the algorithm, all previously generated passwords. Sometimes this is necessary).
Lets you add, delete and list sites for a specific user. Settings for a site can be updated.
Stores a hashed version of a user’s master key on disk in order to tell you when you made a typo in your master password (which would generate the wrong site specific password). scrypt is used for that.
Stores the sites and site settings of a user encrypted. cryptography Fernet is used for that.
Can also be used as a library by other programs.
100% line and branch coverage. Multiple tests to make sure it generates the same passwords as the reference implementation.
Written for Python 3.4 and above.

And these things are planned for the future:

Add import and export functionality ((un)encrypted JSON and the format that the OS X app is currently using).
Allow the generation of login names.
Allow the generation of answers for security questions.
Maybe a GUI using PyQT that runs under OS X, Linux and Windows. But since there are already plenty of GUIs and I don’t have that much time, I may not do it. Please contact me if you are interested in helping me out here.

You can grab your copy of mpw from the Python Package Index. I recommend installing it via pipsi:

$ pipsi install --python `which python3` mpw

But you can also use pip to install it globally or into a virtualenv:

$ pip install mpw

Warning: If you already used mpw 0.2 and have some sites configured, run this script to upgrade you config file. Run it after you installed mpw 0.3 but before actually using it.

SimPy: Real-time simulations

2014-06-25T10:15:00+02:00

This is another guide to SimPy simulations. SimPy ususally runs simulations as fast as possible, but sometimes, you might want your simulation to run synchronized with the wall-clock time. This kind of simulation is also called real-time simulation.

Real-time simulations may be necessary

if you have hardware-in-the-loop,
if there is human interaction with your simulation, or
if you want to analyze the real-time behavior of an algorithm.

To convert a simulation into a real-time simulation, you only need to replace SimPy’s default Environment with a simpy.rt.RealtimeEnvironment. Apart from the initial_time argument, there are two additional parameters: factor and strict: RealtimeEnvironment(initial_time=0, factor=1.0, strict=True).

The factor defines how much real time passes with each step of simulation time. By default, this is one second. If you set factor=0.1, a unit of simulation time will only take a tenth of a second; if you set factor=60, it will take a minute.

Here is a simple example for converting a normal simulation to a real-time simulation with a duration of one tenth of a second per simulation time unit:

>>> import time
>>> import simpy
>>>
>>> def example(env):
...     start = time.perf_counter()
...     yield env.timeout(1)
...     end = time.perf_counter()
...     print('Duration of one simulation time unit: %.2fs' % (end - start))
>>>
>>> env = simpy.Environment()
>>> proc = env.process(example(env))
>>> env.run(until=proc)
Duration of one simulation time unit: 0.00s
>>>
>>> import simpy.rt
>>> env = simpy.rt.RealtimeEnvironment(factor=0.1)
>>> proc = env.process(example(env))
>>> env.run(until=proc)
Duration of one simulation time unit: 0.10s

If the strict parameter is set to True (the default), the step() and run() methods will raise a RuntimeError if the computation within a simulation time step take more time than the real-time factor allows. In the following example, a process will perform a task that takes 0.02 seconds within a real-time environment with a time factor of 0.01 seconds:

>>> import time
>>> import simpy.rt
>>>
>>> def slow_proc(env):
...     time.sleep(0.02)  # Heavy computation :-)
...     yield env.timeout(1)
>>>
>>> env = simpy.rt.RealtimeEnvironment(factor=0.01)
>>> proc = env.process(slow_proc(env))
>>> try:
...     env.run(until=proc)
...     print('Everything alright')
... except RuntimeError:
...     print('Simulation is too slow')
Simulation is too slow

To suppress the error, simply set strict=False:

>>> env = simpy.rt.RealtimeEnvironment(factor=0.01, strict=False)
>>> proc = env.process(slow_proc(env))
>>> try:
...     env.run(until=proc)
...     print('Everything alright')
... except RuntimeError:
...     print('Simulation is too slow')
Everything alright

That’s it. Real-time simulations are that simple with SimPy! This guide is now also part of the official SimPy documentation.

SimPy: Shared Resources

2014-06-13T21:18:00+02:00

Shared resources are another way to model process interaction. They form a congestion point where processes queue up in order to use them.

SimPy defines three categories of resources:

Resources – Resources that can be used by a limited number of processes at a time (e.g., a gas station with a limited number of fuel pumps).
Containers – Resources that model the production and consumption of a homogeneous, undifferentiated bulk. It may either be continuous (like water) or discrete (like apples).
Stores – Resources that allow the production and consumption of Python objects.

The basic concept of resources

All resources share the same basic concept: The resource itself is some kind of a container with a, usually limited, capacity. Processes can either try to put something into the resource or try to get something out. If the resource is full or empty, they have to queue up and wait.

This is roughly, how every resource looks like:

BaseResource(capacity):
   put_queue
   get_queue

   put(): event
   get(): event

Every resources a maximum capacity and two queues, one for processes that want to put something into it and one for processes that want to get something out. The put() and get() methods both return an event that is triggered when the corresponding action was successful.

Resources and interrupts

While a process is waiting for a put or get event to succeed, it may be interrupted by another process. After catching the interrupt, the process has two possibilities:

It may continue to wait for the request (by yielding the event again).
It may stop waiting for the request. In this case, it has to call the event’s cancel() method.

Since you can easily forget this, all resources events are context managers (see the Python docs for details).

The resource system is modular and extensible. Resources can, for example, use specialized queues and event types. This allows them to use sorted queues, to add priorities to events, or to offer preemption.

Resources

Resources can be used by a limited number of processes at a time (e.g., a gas station with a limited number of fuel pumps). Processes request these resources to become a user (or to “own” them) and have to release them once they are done (e.g., vehicles arrive at the gas station, use a fuel-pump, if one is available, and leave when they are done).

Requesting a resources is modeled as “putting a process’ token into the resources” and releasing a resources correspondingly as “getting a process’ token out of the resource”. Thus, calling request()/release() is equivalent to calling put()/get(). Releasing a resource will always succeed immediately.

SimPy implements three resource types:

Resource
PriorityResource, where queueing processes are sorted by priority
PreemptiveResource, where processes additionally may preempt other processes with a lower priority

Resource

The Resource is conceptually a semaphore. Its only parameter – apart from the obligatory reference to an Environment – is its capacity. It must be a positive number and defaults to 1: Resource(env, capacity=1).

Instead of just counting its current users, it stores the request event as an “access token” for each user. This is, for example, useful for adding preemption (see below).

Here is as basic example for using a resource:

>>> import simpy
>>>
>>> def resource_user(env, resource):
...     request = resource.request()  # Generate a request event
...     yield request                 # Wait for access
...     yield env.timeout(1)          # Do something
...     resource.release(request)     # Release the resource
...
>>> env = simpy.Environment()
>>> res = simpy.Resource(env, capacity=1)
>>> user = env.process(resource_user(env, res))
>>> env.run()

Note, that you have to release the resource under all conditions; for example, if you got interrupted while waiting for or using the resource. In order to help you with that and to avoid too many try: ... finally: ... constructs, request events can be used as context manager:

>>> def resource_user(env, resource):
...     with resource.request() as req:  # Generate a request event
...         yield req                    # Wait for access
...         yield env.timeout(1)         # Do something
...                                      # Resource released automatically
>>> user = env.process(resource_user(env, res))
>>> env.run()

Resources allow you retrieve the list of users and queued as well as the number of users and resource’s capacity:

>>> res = simpy.Resource(env, capacity=1)
>>>
>>> def print_stats(res):
...     print('%d of %d slots are allocated.' % (res.count, res.capacity))
...     print('  Users:', res.users)
...     print('  Queued events:', res.queue)
>>>
>>>
>>> def user(res):
...     print_stats(res)
...     with res.request() as req:
...         yield req
...         print_stats(res)
...     print_stats(res)
>>>
>>> procs = [env.process(user(res)), env.process(user(res))]
>>> env.run()
0 of 1 slots are allocated.
  Users: []
  Queued events: []
1 of 1 slots are allocated.
  Users: [<Request() object at 0x...>]
  Queued events: [<Request() object at 0x...>]
1 of 1 slots are allocated.
  Users: [<Request() object at 0x...>]
  Queued events: [<Request() object at 0x...>]
0 of 1 slots are allocated.
  Users: []
  Queued events: [<Request() object at 0x...>]
1 of 1 slots are allocated.
  Users: [<Request() object at 0x...>]
  Queued events: []
0 of 1 slots are allocated.
  Users: []
  Queued events: []

PriorityResource

As you may know from the real world, not every one is equally important. To map that to SimPy, there’s the PriorityResource. This subclass of Resource lets requesting processes provide a priority for each request. More important requests will gain access to the resource earlier than less important ones. Priority is expressed by integer numbers; smaller numbers mean a higher priority.

Apart form that, it works like a normal Resource:

>>> def resource_user(name, env, resource, wait, prio):
...     yield env.timeout(wait)
...     with resource.request(priority=prio) as req:
...         print('%s requesting at %s with priority=%s' % (name, env.now, prio))
...         yield req
...         print('%s got resource at %s' % (name, env.now))
...         yield env.timeout(3)
...
>>> env = simpy.Environment()
>>> res = simpy.PriorityResource(env, capacity=1)
>>> p1 = env.process(resource_user(1, env, res, wait=0, prio=0))
>>> p2 = env.process(resource_user(2, env, res, wait=1, prio=0))
>>> p3 = env.process(resource_user(3, env, res, wait=2, prio=-1))
>>> env.run()
1 requesting at 0 with priority=0
1 got resource at 0
2 requesting at 1 with priority=0
3 requesting at 2 with priority=-1
3 got resource at 3
2 got resource at 6

Although p3 requested the resource later than p2, it could use it earlier because its priority was higher.

PreemptiveResource

Sometimes, new requests are so important that queue-jumping is not enough and they need to kick existing users out of the resource (this is called preemption). The PreemptiveResource allows you to do exactly this:

>>> def resource_user(name, env, resource, wait, prio):
...     yield env.timeout(wait)
...     with resource.request(priority=prio) as req:
...         print('%s requesting at %s with priority=%s' % (name, env.now, prio))
...         yield req
...         print('%s got resource at %s' % (name, env.now))
...         try:
...             yield env.timeout(3)
...         except simpy.Interrupt as interrupt:
...             by = interrupt.cause.by
...             usage = env.now - interrupt.cause.usage_since
...             print('%s got preempted by %s at %s after %s' %
...                   (name, by, env.now, usage))
...
>>> env = simpy.Environment()
>>> res = simpy.PreemptiveResource(env, capacity=1)
>>> p1 = env.process(resource_user(1, env, res, wait=0, prio=0))
>>> p2 = env.process(resource_user(2, env, res, wait=1, prio=0))
>>> p3 = env.process(resource_user(3, env, res, wait=2, prio=-1))
>>> env.run()
1 requesting at 0 with priority=0
1 got resource at 0
2 requesting at 1 with priority=0
3 requesting at 2 with priority=-1
1 got preempted by <Process(resource_user) object at 0x...> at 2 after 2
3 got resource at 2
2 got resource at 5

PreemptiveResource inherits from PriorityResource and adds a preempt flag (that defaults to True) to request(). By setting this to False (resource.request(priority=x, preempt=False)), a process can decide to not preempt another resource user. It will still be put in the queue according to its priority, though.

The implementation of PreemptiveResource values priorities higher than preemption. That means preempt request are not allowed to cheat and jump over a higher prioritized request. The following example shows that preemptive low priority requests cannot queue-jump over high priority requests:

>>> def user(name, env, res, prio, preempt):
...     with res.request(priority=prio, preempt=preempt) as req:
...         try:
...             print('%s requesting at %d' % (name, env.now))
...             yield req
...             print('%s got resource at %d' % (name, env.now))
...             yield env.timeout(3)
...         except simpy.Interrupt:
...             print('%s got preempted at %d' % (name, env.now))
>>>
>>> env = simpy.Environment()
>>> res = simpy.PreemptiveResource(env, capacity=1)
>>> A = env.process(user('A', env, res, prio=0, preempt=True))
>>> env.run(until=1)  # Give A a head start
A requesting at 0
A got resource at 0
>>> B = env.process(user('B', env, res, prio=-2, preempt=False))
>>> C = env.process(user('C', env, res, prio=-1, preempt=True))
>>> env.run()
B requesting at 1
C requesting at 1
B got resource at 3
C got resource at 6

Process A requests the resource with priority 0. It immediately becomes a user.
Process B requests the resource with priority -2 but sets preempt to False. It will queue up and wait.
Process C requests the resource with priority -1 but leaves preempt True. Normally, it would preempt A but in this case, B is queued up before C and prevents C from preempting A. C can also not preempt B since its priority is not high enough.

Thus, the behavior in the example is the same as if no preemption was used at all. Be careful when using mixed preemption!

Due to the higher priority of process B, no preemption occurs in this example. Note that an additional request with a priority of -3 would be able to preempt A.

If your use-case requires a different behaviour, for example queue-jumping or valuing preemption over priorities, you can subclass PreemptiveResource and override the default behaviour.

Containers

Containers help you modelling the production and consumption of a homogeneous, undifferentiated bulk. It may either be continuous (like water) or discrete (like apples).

You can use this, for example, to model the gas / petrol tank of a gas station. Tankers increase the amount of gasoline in the tank while cars decrease it.

The following example is a very simple model of a gas station with a limited number of fuel dispensers (modeled as Resource) and a tank modeled as Container:

>>> class GasStation:
...     def __init__(self, env):
...         self.fuel_dispensers = simpy.Resource(env, capacity=2)
...         self.gas_tank = simpy.Container(env, init=100, capacity=1000)
...         self.mon_proc = env.process(self.monitor_tank(env))
...
...     def monitor_tank(self, env):
...         while True:
...             if self.gas_tank.level < 100:
...                 print('Calling tanker at %s' % env.now)
...                 env.process(tanker(env, self))
...             yield env.timeout(15)
>>>
>>>
>>> def tanker(env, gas_station):
...     yield env.timeout(10)  # Need 10 Minutes to arrive
...     print('Tanker arriving at %s' % env.now)
...     amount = gas_station.gas_tank.capacity - gas_station.gas_tank.level
...     yield gas_station.gas_tank.put(amount)
>>>
>>>
>>> def car(name, env, gas_station):
...     print('Car %s arriving at %s' % (name, env.now))
...     with gas_station.fuel_dispensers.request() as req:
...         yield req
...         print('Car %s starts refueling at %s' % (name, env.now))
...         yield gas_station.gas_tank.get(40)
...         yield env.timeout(5)
...         print('Car %s done refueling at %s' % (name, env.now))
>>>
>>>
>>> def car_generator(env, gas_station):
...     for i in range(4):
...         env.process(car(i, env, gas_station))
...         yield env.timeout(5)
>>>
>>>
>>> env = simpy.Environment()
>>> gas_station = GasStation(env)
>>> car_gen = env.process(car_generator(env, gas_station))
>>> env.run(35)
Car 0 arriving at 0
Car 0 starts refueling at 0
Car 1 arriving at 5
Car 0 done refueling at 5
Car 1 starts refueling at 5
Car 2 arriving at 10
Car 1 done refueling at 10
Car 2 starts refueling at 10
Calling tanker at 15
Car 3 arriving at 15
Car 3 starts refueling at 15
Tanker arriving at 25
Car 2 done refueling at 30
Car 3 done refueling at 30

Containers allow you to retrieve their current level as well as their capacity (see GasStation.monitor_tank() and tanker()). You can also access the list of waiting events via the put_queue and get_queue attributes (similar to Resource.queue).

Stores

Using Stores you can model the production and consumption of concrete objects (in contrast to the rather abstract “amount” stored in containers). A single Store can even contain multiple types of objects.

Beside Store, there is a FilterStore that lets you use a custom function to filter the objects you get out of the store.

Here is a simple example modelling a generic producer/consumer scenario:

>>> def producer(env, store):
...     for i in range(100):
...         yield env.timeout(2)
...         yield store.put('spam %s' % i)
...         print('Produced spam at', env.now)
>>>
>>>
>>> def consumer(name, env, store):
...     while True:
...         yield env.timeout(1)
...         print(name, 'requesting spam at', env.now)
...         item = yield store.get()
...         print(name, 'got', item, 'at', env.now)
>>>
>>>
>>> env = simpy.Environment()
>>> store = simpy.Store(env, capacity=2)
>>>
>>> prod = env.process(producer(env, store))
>>> consumers = [env.process(consumer(i, env, store)) for i in range(2)]
>>>
>>> env.run(until=5)
0 requesting spam at 1
1 requesting spam at 1
Produced spam at 2
0 got spam 0 at 2
0 requesting spam at 3
Produced spam at 4
1 got spam 1 at 4

As with the other resource types, you can get a store’s capacity via the capacity attribute. The attribute items points to the list of items currently available in the store. The put and get queues can be accessed via the put_queue and get_queue attributes.

FilterStore can, for example, be used to model machine shops where machines have varying attributes. This can be useful if the homogeneous slots of a Resource are not what you need:

>>> from collections import namedtuple
>>>
>>> Machine = namedtuple('Machine', 'size, duration')
>>> m1 = Machine(1, 2)  # Small and slow
>>> m2 = Machine(2, 1)  # Big and fast
>>>
>>> env = simpy.Environment()
>>> machine_shop = simpy.FilterStore(env, capacity=2)
>>> machine_shop.items = [m1, m2]  # Pre-populate the machine shop
>>>
>>> def user(name, env, ms, size):
...     machine = yield ms.get(lambda machine: machine.size == size)
...     print(name, 'got', machine, 'at', env.now)
...     yield env.timeout(machine.duration)
...     yield ms.put(machine)
...     print(name, 'released', machine, 'at', env.now)
>>>
>>>
>>> users = [env.process(user(i, env, machine_shop, (i % 2) + 1))
...          for i in range(3)]
>>> env.run()
0 got Machine(size=1, duration=2) at 0
1 got Machine(size=2, duration=1) at 0
1 released Machine(size=2, duration=1) at 1
0 released Machine(size=1, duration=2) at 2
2 got Machine(size=1, duration=2) at 2
2 released Machine(size=1, duration=2) at 4

This guide is now also part of SimPy’s documentation.

SimPy: Process Interaction

2014-04-11T21:08:00+02:00

As promised, the delay between this and the last topical guide was rather short.

This one is about process interaction. This is what makes event discrete simulation interesting. Without it, you actually wouldn’t even need SimPy.

So this guide is about:

sleep-until-woken-up (passivate/reactivate)
waiting-for-another-process-to-terminate
interrupting-another-process

Another possibility for processes to interact are resources. They will be discussed in the next guide.

Sleep until woken up

Imagine you want to model an electric vehicle with an intelligent battery-charging controller. While the vehicle is driving, the controller can be passive but needs to be reactivate once the vehicle is connected to the power grid in order to charge the battery.

In SimPy 2, this pattern was known as passivate / reactivate. In SimPy 3, you can accomplish that with a simple, shared Event:

>>> from random import seed, randint
>>> seed(23)
>>>
>>> import simpy
>>>
>>> class EV:
...     def __init__(self, env):
...         self.env = env
...         self.drive_proc = env.process(self.drive(env))
...         self.bat_ctrl_proc = env.process(self.bat_ctrl(env))
...         self.bat_ctrl_reactivate = env.event()
...
...     def drive(self, env):
...         while True:
...             # Drive for 20-40 min
...             yield env.timeout(randint(20, 40))
...
...             # Park for 1–6 hours
...             print('Start parking at', env.now)
...             self.bat_ctrl_reactivate.succeed()  # "reactivate"
...             self.bat_ctrl_reactivate = env.event()
...             yield env.timeout(randint(60, 360))
...             print('Stop parking at', env.now)
...
...     def bat_ctrl(self, env):
...         while True:
...             print('Bat. ctrl. passivating at', env.now)
...             yield self.bat_ctrl_reactivate  # "passivate"
...             print('Bat. ctrl. reactivated at', env.now)
...
...             # Intelligent charging behavior here …
...             yield env.timeout(randint(30, 90))
...
>>> env = simpy.Environment()
>>> ev = EV(env)
>>> env.run(until=150)
Bat. ctrl. passivating at 0
Start parking at 29
Bat. ctrl. reactivated at 29
Bat. ctrl. passivating at 60
Stop parking at 131

Since bat_ctrl() just waits for a normal event, we no longer call this pattern passivate / reactivate in SimPy 3.

Waiting for another process to terminate

The example above has a problem: it may happen that the vehicles wants to park for a shorter duration than it takes to charge the battery (this is the case if both, charging and parking would take 60 to 90 minutes).

To fix this problem we have to slightly change our model. A new bat_ctrl() will be started every time the EV starts parking. The EV then waits until the parking duration is over and until the charging has stopped:

>>> class EV:
...     def __init__(self, env):
...         self.env = env
...         self.drive_proc = env.process(self.drive(env))
...
...     def drive(self, env):
...         while True:
...             # Drive for 20-40 min
...             yield env.timeout(randint(20, 40))
...
...             # Park for 1–6 hours
...             print('Start parking at', env.now)
...             charging = env.process(self.bat_ctrl(env))
...             parking = env.timeout(randint(60, 360))
...             yield charging & parking
...             print('Stop parking at', env.now)
...
...     def bat_ctrl(self, env):
...         print('Bat. ctrl. started at', env.now)
...         # Intelligent charging behavior here …
...         yield env.timeout(randint(30, 90))
...         print('Bat. ctrl. done at', env.now)
...
>>> env = simpy.Environment()
>>> ev = EV(env)
>>> env.run(until=310)
Start parking at 29
Bat. ctrl. started at 29
Bat. ctrl. done at 83
Stop parking at 305

Again, nothing new (if you’ve read the events guide) and special is happening. SimPy processes are events, too, so you can yield them and will thus wait for them to get triggered. You can also wait for two events at the same time by concatenating them with &.

Interrupting another process

As usual, we now have another problem: Imagine, a trip is very urgent, but with the current implementation, we always need to wait until the battery is fully charged. If we could somehow interrupt that …

Fortunate coincidence, there is indeed a way to do exactly this. You can call interrupt() on a Process. This will throw an Interrupt exception into that process, resuming it immediately:

>>> class EV:
...     def __init__(self, env):
...         self.env = env
...         self.drive_proc = env.process(self.drive(env))
...
...     def drive(self, env):
...         while True:
...             # Drive for 20-40 min
...             yield env.timeout(randint(20, 40))
...
...             # Park for 1 hour
...             print('Start parking at', env.now)
...             charging = env.process(self.bat_ctrl(env))
...             parking = env.timeout(60)
...             yield charging | parking
...             if not charging.triggered:
...                 # Interrupt charging if not already done.
...                 charging.interrupt('Need to go!')
...             print('Stop parking at', env.now)
...
...     def bat_ctrl(self, env):
...         print('Bat. ctrl. started at', env.now)
...         try:
...             yield env.timeout(randint(60, 90))
...             print('Bat. ctrl. done at', env.now)
...         except simpy.Interrupt as i:
...             # Onoes! Got interrupted before the charging was done.
...             print('Bat. ctrl. interrupted at', env.now, 'msg:',
...                   i.cause)
...
>>> env = simpy.Environment()
>>> ev = EV(env)
>>> env.run(until=100)
Start parking at 31
Bat. ctrl. started at 31
Stop parking at 91
Bat. ctrl. interrupted at 91 msg: Need to go!

What process.interrupt() actually does is removing the process’ _resume() method from the callbacks of the event that it is currently waiting for. And it will schedule an event that will throw the Interrupt exception into the interrupted process as soon as possible.

Since we don’t to anything special to the event, the interrupted process can yield the same event again after catching the Interrupt – Imagine someone waiting for a shop to open. The person may get interrupted by a phone call. After finishing the call, he or she checks if the shop already opened and either enters or continues to wait.

This guide is now, of course, also part of the SimPy documentation.

SimPy: Events

2014-04-05T11:52:00+02:00

It’s been a while since the last SimPy guide, but I’ve been quite busy with developing and Open Sourcing mosaik 2.

“What’s SimPy again?” SimPy is a process-based discrete-event simulation framework based on standard Python. Its event dispatcher is based on Python’s generators and can also be used for asynchronous networking or to implement multi-agent systems (with both, simulated and real communication).

The first to guides provided a general description of how SimPy works and an explanation of SimPy environments.

This guide is all about events.

SimPy includes an extensive set of event types for various purposes. All of them inherit simpy.events.Event. The listing below shows the hierarchy of events built into SimPy:

events.Event
↑
+— events.Timeout
|
+— events.Initialize
|
+— events.Process
|
+— events.Condition
|  ↑
|  +— events.AllOf
|  |
|  +— events.AnyOf
⋮
+– [resource events]

This is the set of basic events. Events are extensible and resources, for example, define additional events. In this guide, we’ll focus on the events in the simpy.events module.

Event basics

SimPy events are very similar – if not identical — to deferreds, futures or promises. Instances of the class Event are used to describe any kind of events. Events can be in one of the following states. An event

might happen (not triggered),
is going to happen (triggered) or
has happened (processed).

They traverse these states exactly once in that order. Events are also tightly bound to time and time causes events to advance their state.

Initially, events are not triggered and just objects in memory.

If an event gets triggered, it is scheduled at a given time and inserted into SimPy’s event queue. The property Event.triggered becomes True.

As long as the event is not processed, you can add callbacks to an event. Callbacks are callables that accept an event as parameter and are stored in the Event.callbacks list.

An event becomes processed when SimPy pops it from the event queue and calls all of its callbacks. It is now no longer possible to add callbacks. The property Event.processed becomes True.

Events also have a value. The value can be set before or when the event is triggered and can be retrieved via Event.value or, within a process, by yielding the event (value = yield event).

Adding callbacks to an event

“What? Callbacks? I’ve never seen no callbacks!”, you might think if you have worked your way through the tutorial.

That’s on purpose. The most common way to add a callback to an event is yielding it from your process function (yield event). This will add the process’ _resume() method as a callback. That’s how your process gets resumed when it yielded an event.

However, you can add any callable object (function) to the list of callbacks as long as it accepts an event instance as its single parameter:

>>> import simpy
>>>
>>> def my_callback(event):
...     print('Called back from', event)
...
>>> env = simpy.Environment()
>>> event = env.event()
>>> event.callbacks.append(my_callback)
>>> event.callbacks
[<function my_callback at 0x...>]

If an event has been processed, all of its Event.callbacks have been executed and the attribute is set to None. This is to prevent you from adding more callbacks – these would of course never get called because the event has already happened.

Processes are smart about this, though. If you yield a processed event, _resume() will immediately resume your process with the value of the event (because there is nothing to wait for).

Triggering events

When events are triggered, they can either succeed or fail. For example, if an event is to be triggered at the end of a computation and everything works out fine, the event will succeed. If an exceptions occurs during that computation, the event will fail.

To trigger an event and mark it as successful, you can use Event.succeed(value=None). You can optionally pass a value to it (e.g., the results of a computation).

To trigger an event and mark it as failed, call Event.fail(exception) and pass an Exception instance to it (e.g., the exception you caught during your failed computation).

There is also a generic way to trigger an event: Event.trigger(event). This will take the value and outcome (success or failure) of the event passed to it.

All three methods return the event instance they are bound to. This allows you to do things like yield Event(env).succeed().

Example usages for `Event`

The simple mechanics outlined above provide a great flexibility in the way events (even the basic Event) can be used.

One example for this is that events can be shared. They can be created by a process or outside of the context of a process. They can be passed to other processes and chained:

>>> class School:
...     def __init__(self, env):
...         self.env = env
...         self.class_ends = env.event()
...         self.pupil_procs = [env.process(self.pupil()) for i in range(3)]
...         self.bell_proc = env.process(self.bell())
...
...     def bell(self):
...         for i in range(2):
...             yield self.env.timeout(45)
...             self.class_ends.succeed()
...             self.class_ends = self.env.event()
...             print()
...
...     def pupil(self):
...         for i in range(2):
...             print(' \o/', end='')
...             yield self.class_ends
...
>>> school = School(env)
>>> env.run()
 \o/ \o/ \o/
 \o/ \o/ \o/

This can also be used like the passivate / reactivate known from SimPy 2. The pupils passivate when class begins and are reactivated when the bell rings.

Let time pass by: the `Timeout`

To actually let time pass in a simulation, there is the timeout event. A timeout has two parameters: a delay and an optional value: Timeout(delay, value=None). It triggers itself during its creation and schedules itself at now + delay. Thus, the succeed() and fail() methods cannot be called again and you have to pass the event value to it when you create the timeout.

The delay can be any kind of number, usually an int or float as long as it supports comparison and addition.

Processes are events, too

SimPy processes (as created by Process or env.process()) have the nice property of being events, too.

That means, that a process can yield another process. It will then be resumed when the other process ends. The event’s value will be the return value of that process:

>>> def sub(env):
...     yield env.timeout(1)
...     return 23
...
>>> def parent(env):
...     ret = yield env.process(sub(env))
...     return ret
...
>>> env.run(env.process(parent(env)))
23

The example above will only work in Python >= 3.3. As a workaround for older Python versions, you can use env.exit(23) with the same effect.

When a process is created, it schedules an Initialize event which will start the execution of the process when triggered. You usually won’t have to deal with this type of event.

If you don’t want a process to start immediately but after a certain delay, you can use simpy.util.start_delayed(). This method returns a helper process that uses a timeout before actually starting a process.

The example from above, but with a delayed start of sub():

>>> from simpy.util import start_delayed
>>>
>>> def sub(env):
...     yield env.timeout(1)
...     return 23
...
>>> def parent(env):
...     start = env.now
...     sub_proc = yield start_delayed(env, sub(env), delay=3)
...     assert env.now - start == 3
...
...     ret = yield sub_proc
...     return ret
...
>>> env.run(env.process(parent(env)))
23

Waiting for for multiple events at once

Sometimes, you want to wait for more than one event at the same time. For example, you may want to wait for a resource, but not for an unlimited amount of time. Or you may want to wait until all of a number of events have happened.

SimPy therefore offers the AnyOf and AllOf events which both are a Condition event.

Both take a list of events as an argument and are triggered if at least one of them is triggered or all of them:

>>> from simpy.events import AnyOf, AllOf, Event
>>> events = [Event(env) for i in range(3)]
>>> a = AnyOf(env, events)  # Triggers if at least one of "events" is triggered.
>>> b = AllOf(env, events)  # Triggers if all each of "events" is triggered.

The value of a condition event is a dictionary with an entry for every triggered event. In the case of AllOf, the size of that dictionary will always be the same as the length of the event list. The value dict of AnyOf will have at least one entry. In both cases, the event instances are used as keys and the event values will be the values.

As a shorthand for AllOf and AnyOf, you can also use the logical operators & (and) and | (or):

>>> def test_condition(env):
...     t1, t2 = env.timeout(1, value='spam'), env.timeout(2, value='eggs')
...     ret = yield t1 | t2
...     assert ret == {t1: 'spam'}
...
...     t1, t2 = env.timeout(1, value='spam'), env.timeout(2, value='eggs')
...     ret = yield t1 & t2
...     assert ret == {t1: 'spam', t2: 'eggs'}
...
...     # You can also concatenate & and |
...     e1, e2, e3 = [env.timeout(i) for i in range(3)]
...     yield (e1 | e2) & e3
...     assert all(e.triggered for e in [e1, e2, e3])
...
>>> proc = env.process(test_condition(env))
>>> env.run()

That’s it. I’ve already started with the next guide – about process interaction – so hopefully it won’t take to long until publish it this time.

mosaik – An open co-simulation framework for smart energy systems

2014-03-27T13:37:00+01:00

Since nearly four years, we’ve been developing and working with mosaik within our research group at OFFIS. Now, we finally had the opportunity to release it as Open Source Software.

You can read the full announcement here.

SimPy: Environments

2014-01-20T17:33:00+01:00

This is the second in a series of guides that describe how SimPy works and how to use it best. This time I’ll discuss environments.

A simulation environment manages the simulation time as well as the scheduling and processing of events. It also provides means to step through or execute the simulation.

The base class for all environments is BaseEnvironment. “Normal” simulations usually use its subclass Environment. For real-time simulations, SimPy provides a RealtimeEnvironment (more on that in another guide).

Simulation control

SimPy is very flexible in terms of simulation execution. You can run your simulation until there is no more event, until a certain simulation time is reached, or until a certain event is triggered. You can also step through the simulation event by event. Furthermore, you can mix these things as you like.

For example, you could run your simulation until an interesting event occurs. You could then step through the simulation event by event for a while; and finally run the simulation until there is no more event left and your processes all have terminated.

The most important method here is Environment.run():

If you call it without any argument (env.run()), it steps through the simulation until there is no more event left.

WARNING: If your processes run forever (while True: yield env.timeout(1)), this method will never terminate (unless you kill your script by e.g., pressing Ctrl-C).
In most cases it is more advisable to stop your simulation when it reaches a certain simulation time. Therefore, you can pass the desired time via the until parameter, e.g.: env.run(until=10).

The simulation will then stop when the internal clock reaches 10 but will not process any events scheduled for time 10. This is similar to a new environment where the clock is 0 but (obviously) no events have yet been processed.

If you want to integrate your simulation in a GUI and want to draw a process bar, you can repeatedly call this function with increasing until values and update your progress bar after each call:
```
for i in range(100):
    env.run(until=i)
    progressbar.update(i)
```
Instead of passing a number to run(), you can also pass any event to it. run() will then return when the event has been processed.

Assuming that the current time is 0, env.run(until=env.timeout(5)) is equivalent to env.run(until=5).

You can also pass other types of events (remember, that a Process is an event, too):
```
>>> import simpy
>>>
>>> def my_proc(env):
...     yield env.timeout(1)
...     return 'Monty Python’s Flying Circus'
>>>
>>> env = simpy.Environment()
>>> proc = env.process(my_proc(env))
>>> env.run(until=proc)
'Monty Python’s Flying Circus'
```

To step through the simulation event by event, the environment offers peek() and step().

peek() returns the time of the next scheduled event of infinity (float('inf')) of no more event is scheduled.

step() processes the next scheduled event. It raises an EmptySchedule exception if no event is available.

In a typical use case, you use these methods in a loop like:

until = 10
while env.peek() < until:
   env.step()

State access

The environment allows you to get the current simulation time via the Environment.now property. The simulation time is a number without unit and is increased via Timeout events.

By default, now starts at 0, but you can pass an initial_time to the Environment to use something else.

NOTE: Although the simulation time is technically unitless, you can pretend that it is, for example, in seconds and use it like a timestamp returned by time.time() to calculate a date or the day of the week.

The property Environment.active_process is comparable to os.getpid() and is either None or pointing at the currently active Process. A process is active when its process function is being executed. It becomes inactive (or suspended) when it yields an event.

Thus, it makes only sense to access this property from within a process function or a function that is called by your process function:

>>> def subfunc(env):
...     print(env.active_process)  # will print "p1"
>>>
>>> def my_proc(env):
...     while True:
...         print(env.active_process)  # will print "p1"
...         subfunc(env)
...         yield env.timeout(1)
>>>
>>> env = simpy.Environment()
>>> p1 = env.process(my_proc(env))
>>> env.active_process  # None
>>> env.step()
<Process(my_proc) object at 0x...>
<Process(my_proc) object at 0x...>
>>> env.active_process  # None

An exemplary use case for this is the resource system: If a process function calls request() to request a resource, the resource determines the requesting process via env.active_process. Take a look at the code to see how we do this :-).

Event creation

To create events, you normally have to import simpy.events, instantiate the event class and pass a reference to the environment to it. To reduce the amount of typing, the Environment provides some shortcuts for event creation. For example, Environment.event() is equivalent to simpy.events.Event(env).

Other shortcuts are:

More details on what the events do can be found in the guide to events (not yet written :-)).

Miscellaneous

Since Python 3.3, a generator function can have a return value:

def my_proc(env):
    yield env.timeout(1)
    return 42

In SimPy, this can be used to provide return values for processes that can be used by other processes:

def other_proc(env):
    ret_val = yield env.process(my_proc(env))
    assert ret_val == 42

Internally, Python passes the return value as parameter to the StopIteration exception that it raises when a generator is exhausted. So in Python 2.7 and 3.2 you could replace the return 42 with a raise StopIteration(42) to achieve the same result.

To keep your code more readable, the environment provides the method exit() to do exactly this:

def my_proc(env):
    yield env.timeout(1)
    env.exit(42)  # Py2 equivalent to "return 42"

You can find the complete guide on Read the Docs. The next one will be about events and the event types provided by SimPy.

How SimPy works

2013-12-03T09:48:00+01:00

SimPy is a discrete event-simulation library for Python. Recently, the new version 3 has ben released. With that new release, we also changed the structure of the documentation. It now features a Tutorial, Topical Guides, an API Reference and a list of examples. Most of the topical guides have yet to written, though. So this is the first in a series of posts describing various concepts of SimPy.

If you break SimPy down, it is just an asynchronous event dispatcher. You generate events and schedule them at a given simulation time. Events are sorted by priority, simulation time, and an increasing event id. An event also has a list of callbacks, which are executed when the event is triggered and processed by the event loop. Events may also have a return value.

The components involved in this are the Environment, events and the process functions that you write.

Process functions implement your simulation model, that is, they define the behavior of your simulation. They are plain Python generator functions that yield instances of Event.

The environment stores these events in its event list and keeps track of the current simulation time.

If a process function yields and event, SimPy adds the process to the event’s callbacks and suspends the process until the event is triggered and processed. When a process waiting for an event is resumed, it will also receive the event’s value.

Here is a very simple example that illustrates all this; the code is more verbose than it needs to be to make things extra clear. You find a compact version of it at the end of this section:

>>> import simpy
>>>
>>> def example(env):
...     event = simpy.events.Timeout(env, delay=1, value=42)
...     value = yield event
...     print('now=%d, value=%d' % (env.now, value))
>>>
>>> env = simpy.Environment()
>>> example_gen = example(env)
>>> p = simpy.events.Process(env, example_gen)
>>>
>>> env.run()
now=1, value=42

The example() process function above first creates a Timeout event. It passes the environment, a delay, and a value to it. The Timeout schedules itself at now + delay (that’s why the environment is required); other event types usually schedule themselves at the current simulation time.

The process function then yields the event and thus gets suspended. It is resumed, when SimPy processes the Timeout event. The process function also receives the event’s value (42) – this is, however, optional, so yield event would have been okay if the you were not interested in the value or if the event had no value at all.

Finally, the process function prints the current simulation time (that is accessible via the environment’s now attribute) and the Timeout’s value.

If all required process functions are defined, you can instantiate all objects for your simulation. In most cases, you start by creating an instance of Environment, because you’ll need to pass it around a lot when creating everything else.

Starting a process function involves two things:

You have to call the process function to create a generator object. (This will not execute any code of that function yet. Please read The Python yield keyword explained, to understand why this is the case.)
You then create an instance of Process and pass the environment and the generator object to it. This will schedule an Initialize event at the current simulation time which starts the execution of the process function. The process instance is also an event that is triggered when the process function returns. The guide to events (not yet written) explains why this is handy.

Finally, you can start SimPy’s event loop. By default, it will run as long as there are events in the event list, but you can also let it stop earlier by providing an until argument.

The next guide/post will describe the environment and its interactions with events and process functions in more detail.

“Best practice” version of the example above

>>> import simpy
>>>
>>> def example(env):
...     value = yield env.timeout(1, value=42)
...     print('now=%d, value=%d' % (env.now, value))
>>>
>>> env = simpy.Environment()
>>> p = env.process(example(env))
>>> env.run()
now=1, value=42

SimPy 3.0.2 released

2013-10-24T22:37:00+02:00

SimPy 3.0.2 has just been released. It fixes the default capacity for Container, Store and FilterStore, which is now unlimited.

What is SimPy?

SimPy is a process-based discrete-event simulation framework based on standard Python. Its event dispatcher is based on Python’s generators and can also be used for asynchronous networking or to implement multi-agent systems (with both, simulated and real communication).

SimPy 3 has been rewritten from scratch. This is the original release announcement:

After one and a half year of development and many iterations, prototypes, and endless discussions, we now proudly release SimPy 3.

SimPy 3 has been completely rewritten from scratch. Our main goals were to simplify the API and code base as well as making SimPy more flexible and extensible. Some of the most important changes are:

Stronger focus on events. Processes yield event instances and are suspended until the event is triggered. An example for an event is a timeout (formerly known as hold), but even processes are now events, too (you can wait until a process terminates).

Events can be combined with & (and) and | (or) to create condition events.

Process can now be defined by any generator function. You don’t have to subclass Process anymore.

No more global simulation state. Every simulation stores its state in an environment which is comparable to the old Simulation class.

Improved resource system with newly added resource types.

Removed plotting and GUI capabilities. Pyside and matplotlib are much better with this.

Greatly improved test suite. Its cleaner, and the tests are shorter and more numerous.

Completely overhauled documentation.

There is a guide for porting from SimPy 2 to SimPy 3. If you want to stick to SimPy 2 for a while, change your requirements to 'SimPy>=2.3,<3'.

All in all, SimPy has become a framework for asynchronous programming based on coroutines. It brings more than ten years of experience and scientific know-how in the field of event-discrete simulation to the world of asynchronous programming and should thus be a solid foundation for everything based on an event loop.

SimPy 3 is dedicated to Klaus G. Müller and Tony Vignaux who originally created SimPy in 2002. Without their hard work and efforts to create and document SimPy and to build a friendly community around it, we wouldn’t be where we are today. Thank you!

SimPy 3 released

2013-10-11T22:35:00+02:00

After one and a half year of development and many iterations, prototypes, and endless discussions, we now proudly release SimPy 3.

SimPy 3 has been completely rewritten from scratch. Our main goals were to simplify the API and code base as well as making SimPy more flexible and extensible. Some of the most important changes are:

Stronger focus on events. Processes yield event instances and are suspended until the event is triggered. An example for an event is a timeout (formerly known as hold), but even processes are now events, too (you can wait until a process terminates).
Events can be combined with & (and) and | (or) to create condition events.
Process can now be defined by any generator function. You don’t have to subclass Process anymore.
No more global simulation state. Every simulation stores its state in an environment which is comparable to the old Simulation class.
Improved resource system with newly added resource types.
Removed plotting and GUI capabilities. Pyside and matplotlib are much better with this.
Greatly improved test suite. Its cleaner, and the tests are shorter and more numerous.
Completely overhauled documentation.

There is a guide for porting from SimPy 2 to SimPy 3. If you want to stick to SimPy 2 for a while, change your requirements to 'SimPy>=2.3,<3'.

All in all, SimPy has become a framework for asynchronous programming based on coroutines. It brings more than ten years of experience and scientific know-how in the field of event-discrete simulation to the world of asynchronous programming and should thus be a solid foundation for everything based on an event loop.

SimPy 3 is dedicated to Klaus G. Müller and Tony Vignaux who originally created SimPy in 2002. Without their hard work and efforts to create and document SimPy and to build a friendly community around it, we wouldn’t be where we are today. Thank you!

Handling sub-process hierarchies in Python on Linux, OS X and Windows

2013-08-15T12:52:34+02:00

TL;DR—Windows doesn’t support Posix signals. I’ll show you how you can work around this to cleanly terminate process hierarchies. Take a look at the final example.

In this article, I’m trying solve an—at a first glance—simple problem:

You have a Python process that starts one ore more sub-processes, which again might start their own sub-processes. If you press Ctrl-C on the command line, your main process and all of its sub-process should stop, maybe performing some clean-up before they terminate. Also, one process may decide to stop one of its children (including all of its sub-processes).

Example: You have a server process that you start via the command line and that starts several sub-processes. You usually stop that server and all sub-processes by hitting Ctrl-C.

You also want to start your server from within a test case to test its replies on certain requests. If the test is done, you have to send some kind of a message or signal to the server to stop it and all of its sub-processes. The message should obviously be nothing that normal users can send to the server.

This should not only work on Linux and OS X (which would be trivial), but also on Windows.

Before I’ll show you some code, I’ll briefly describe what happens when you press Ctrl-C in a shell and which types of inter-process communication exist to tell another process to stop. After that, I’ll introduce a simple example which I’m then going to extend until the requirements stated above are met.

I’m using Python 3.3 for all examples. The code is tested on Ubuntu 13.04, OS X 10.8 and Windows 7.

Signals and inter-process communication

Signals are a simple form of inter-process communication (IPC) on Posix-compliant operating systems (like Linux or OS X). If one processes sends a signal to another one, the OS interrupts its execution to deliver the signal. The receiving process may now handle the signal.

Some interesting signals for this article are (quoted from Wikipedia):

SIGINT: “The SIGINT signal is sent to a process by its controlling terminal when a user wishes to interrupt the process. This is typically initiated by pressing Ctrl-C, but on some systems, the ‘delete’ character or ‘break’ key can be used.”
SIGTERM: “The SIGTERM signal is sent to a process to request its termination. Unlike the SIGKILL signal, it can be caught and interpreted or ignored by the process. This allows the process to perform nice termination releasing resources and saving state if appropriate. It should be noted that SIGINT is nearly identical to SIGTERM.”
SIGKILL: “The SIGKILL signal is sent to a process to cause it to terminate immediately. In contrast to SIGTERM and SIGINT, this signal cannot be caught or ignored, and the receiving process cannot perform any clean-up upon receiving this signal.”

Though SIGINT and SIGTERM are quite similar, the difference between them is from my point of view that SIGINT is something directly initiated by the user (you press Ctrl-C in your terminal) while SIGTERM is rather used programmatically (e.g., processes receive a SIGTERM when the OS is shutting down).

You can also group processes into process groups and send a signal to all processes in that group at once. That’s for example what your terminal does when you press Ctrl-C.

If a Python program receives a SIGINT, the default handler raises a KeyboardInterrupt by default. You can catch that exception and handle the interrupt in anyway you want (e.g., terminate immediately, do some clean-up before or just ignore it). Martin Cracauer discusses how a SIGINT should be handled properly.

To handle a SIGTERM or suppress the KeyboardInterrupt exception in Python, you can register your own handler via signal.signal().

Since Windows is not Posix-compliant (what a shame), most of the signals won’t work there, but we’ll come to this later. Tim Golden also wrote about this. Unix process groups also don’t work on Windows.

Other, more complicated (and more powerful) ways for IPC are, for example, files, pipes and sockets. Using these, you can just send a message like “plzdiekthxbye” to another process to tell it to stop.

Stopping processes via Ctrl-C

For all examples, we’ll use a simple process hierarchy: Process A starts process B and process B starts process C. Every process will sleep for ten seconds and waits for a KeyboardInterrupt:

import subprocess
import sys
import time


PYTHON - sys.executable
SCRIPT - __file__


def main(name):
    print('%s started' % name)

    # A and B spawn a subprocess
    if name -- 'A':
        child - subproc('B')
    elif name -- 'B':
        child - subproc('C')
    else:
        child - None

    # Sleep and wait for a Ctrl-C
    try:
        time.sleep(10)
        print('%s done' % name)
    except KeyboardInterrupt:
        print('%s got KeyboardInterrupt' % name)
    finally:
        if child:
            child.wait()


def subproc(name):
    """Create and return a new subprocess named *name*."""
    proc - subprocess.Popen([PYTHON, SCRIPT, name])
    return proc


if __name__ -- '__main__':
    name - sys.argv[1] if len(sys.argv) > 1 else 'A'
    main(name)

Running this script and pressing Ctrl-C after about one or two seconds gives us the following output:

$ python subprocs_1.py
A started
B started
C started
^CA got KeyboardInterrupt
B got KeyboardInterrupt
C got KeyboardInterrupt

So far so good. It will get more complicated soon.

Stopping a sub-process and all of its children

To request the termination of a subprocess, we’ll send a SIGTERM to it (on Linux and OS X, this signal is sent if you call a Popen object’s terminate() method). For the process being terminated, this means the same as a SIGINT—it is allowed to do some clean-up or even to ignore the signal.

To catch the signal, you have to register a custom handler, though. To make things easier, you should also register the same handler for the SIGINT signal, so that you don’t end up having the same code in the handler function and in an except KeyboardInterrupt block:

import functools
import signal
import subprocess
import sys
import time
import traceback


PYTHON - sys.executable
SCRIPT - __file__
SIGNALS - {
    signal.SIGINT: 'SIGINT',
    signal.SIGTERM: 'SIGTERM',
}


def main(name, terminate):
    """If *terminate* is ``True`` (should only be the case if *name* is ``A``),
    A will try to terminate B.

    B and C will always just sleep and wait for things to happen ...

    """
    print('%s started' % name)

    # A and B spawn a subprocess
    if name -- 'A':
        child - subproc('B')
    elif name -- 'B':
        child - subproc('C')
    else:
        child - None

    # Curry our cleanup func and register it as handler for SIGINT and SIGTERM
    handler - functools.partial(cleanup, name, child)
    signal.signal(signal.SIGINT, handler)
    signal.signal(signal.SIGTERM, handler)

    if terminate:
        # A tries to terminate B
        time.sleep(1)
        term(child)
        print('%s ended' % name)
    else:
        time.sleep(10)
        print('%s done' % name)
        if child:
            child.wait()


def subproc(name):
    """Create and return a new subprocess named *name*."""
    proc - subprocess.Popen([PYTHON, SCRIPT, name])
    return proc


def term(proc):
    """Send a SIGTERM to *proc* and wait for it to terminate."""
    proc.terminate()  # Sends SIGTERM
    proc.wait()


def cleanup(name, child, signum, frame):
    """Stop the sub-process *child* if *signum* is SIGTERM. Then terminate."""
    try:
        print('%s got a %s' % (name, SIGNALS[signum]))
        if child and signum != signal.SIGINT:
            term(child)
    except:
        traceback.print_exc()
    finally:
        sys.exit()


if __name__ -- '__main__':
    terminate - False
    if len(sys.argv) -- 1:
        name - 'A'
    elif sys.argv[1] -- 'term':
        terminate - True
        name - 'A'
    else:
        name - sys.argv[1]  # B or C

    main(name, terminate)

Our processes now handle SIGINT and SIGTERM signals via the cleanup() handler. The curried version of this handler will have the name of the process and its child process set by default, leaving its signature as handler(signum, frame) which is exactly what signal.signal() expects. Note, that we don’t need to terminate our sub-processes when we get a SIGINT, since the sub-processes will also receive one.

We can now also pass a terminate argument to our script. A will then try to terminate B after a second. B will then stop C.

If you run this script without arguments, you will get (nearly) the same output as in the last example. If you pass term, you’ll get the following output from a Linux or OS X shell:

$ python subprocs_2.py term
A started
B started
C started
B got a SIGTERM
C got a SIGTERM
A ended

If you run the same thing on Windows, you’ll get:

>python subprocs_2.py term
A started
B started
C started
A ended

>C done

Our code now magically stopped working on Windows :-). B seems to receive something from A, but immediately terminates (no output is printed). It also does not forward the signal to C, so C just waits ten seconds and prints its message before exiting on its own.

Clearly not the desired behavior, so this is where the fun begins …

Fixing it on Windows

If you search for this problem on the Interwebs, you’ll find a lot of different answers—some more helpful, some less. This Reddit post helped me the most. According to the MSDN, you can use GenerateConsoleCtrlEvent to send two types of signals to a process: CTRL_C_EVENT and CTRL_BREAK_EVENT. The former translates to a SIGINT, the latter to a SIGBREAK. Instead of GenerateConsoleCtrlEvent, which is only available via the win32 API, you can fortunately also use os.kill or Popen.send_signal() to send the signal.

A CTRL_C_EVENT can not be directed to a single process and is always received by all processes that share the current console. CTRL_BREAK_EVENT on the other hand can be send to specific processes. However, in order to use it, we have to pass creationgflags-subprocess.CREATE_NEW_PROCESS_GROUP to the Popen constructor. This parameter is only available on Windows and has nothing to do with Unix process groups. When you set this flag, you can no longer send CTRL_C_EVENT and if you press Ctrl-C in your console, only the root process will receive a SIGINT signal. Also note, that sending any other signal than CTRL_C_EVENT and CTRL_BREAK_EVENT will unconditionally kill the process.

We’ll end up with the following preconditions:

If on Linux or OS X:

Your terminal sends a SIGINT to all processes if you press Ctrl-C. There’s no need to forward it to sub-processes.
If a process receives a SIGTERM, it should forward it to its sub-processes.

If on Windows:

You should start sub-processes with creationflags-CREATE_NEW_PROCESS_GROUP
This enables you to send a CTRL_BREAK_EVENT to a specific process. The process receives a SIGBREAK and should forward it to its sub-processes.
If you press Ctrl-C in your console, only the root process receives a SIGINT and should send a CTRL_BREAK_EVENT to its sub-processes.

Incorporating what we know now, our example looks like this:

import functools
import signal
import subprocess
import sys
import time
import traceback


PYTHON - sys.executable
SCRIPT - __file__
ON_WINDOWS - (sys.platform -- 'win32')
SIGNALS - {
    signal.SIGINT: 'SIGINT',
    signal.SIGTERM: 'SIGTERM',
}
if ON_WINDOWS:
    SIGNALS[signal.SIGBREAK] - 'SIGBREAK'


def main(name, terminate):
    """If *terminate* is ``True`` (should only be the case if *name* is ``A``),
    A will try to terminate B.

    B and C will always just sleep and wait for things to happen ...

    """
    print('%s started' % name)

    # A and B spawn a subprocess
    if name -- 'A':
        child - subproc('B')
    elif name -- 'B':
        child - subproc('C')
    else:
        child - None

    # Curry our cleanup func and register it as handler for SIGINT and SIGTERM
    handler - functools.partial(cleanup, name, child)
    signal.signal(signal.SIGINT, handler)
    if ON_WINDOWS:
        signal.signal(signal.SIGBREAK, handler)
    else:
        signal.signal(signal.SIGTERM, handler)

    if terminate:
        # A tries to terminate B
        time.sleep(1)
        term(child)
        print('%s ended' % name)
    else:
        # SIGBREAK cannot interrupt sleep(), so we sleep 10 * 1s instead
        for i in range(10):
            time.sleep(1)
        print('%s done' % name)
        if child:
            child.wait()


def subproc(name):
    """Create and return a new subprocess named *name*."""
    kwargs - {}
    if ON_WINDOWS:
        kwargs['creationflags'] - subprocess.CREATE_NEW_PROCESS_GROUP
    proc - subprocess.Popen([PYTHON, SCRIPT, name], **kwargs)
    return proc


def term(proc):
    """Send a SIGTERM/SIGBREAK to *proc* and wait for it to terminate."""
    if ON_WINDOWS:
        proc.send_signal(signal.CTRL_BREAK_EVENT)
    else:
        proc.terminate()
    proc.wait()


def cleanup(name, child, signum, frame):
    """Stop the sub-process *child* if *signum* is SIGTERM. Then terminate."""
    try:
        print('%s got a %s' % (name, SIGNALS[signum]))
        if child and (ON_WINDOWS or signum != signal.SIGINT):
            # Forward SIGTERM on Linux or any signal on Windows
            term(child)
    except:
        traceback.print_exc()
    finally:
        sys.exit()


if __name__ -- '__main__':
    terminate - False
    if len(sys.argv) -- 1:
        name - 'A'
    elif sys.argv[1] -- 'term':
        terminate - True
        name - 'A'
    else:
        name - sys.argv[1]  # B or C

    main(name, terminate)

We can register a SIGINT handler in any case. If on Windows we also register a SIGBREAK handler. For all remaining cases we register a handler for SIGTERM.

Since SIGBREAK cannot interrupt a time.sleep() call, I changed it from time.sleep(10) to for i in range(10): time.sleep(1) so that the sub-processes terminate faster. I don’t know if this is intended behavior or a bug in Python.

When terminating a process, you have to send a CTRL_BREAK_EVENT on Windows. Else, you can just use Popen.terminate() which will send a SIGTERM to the process.

In our clean-up function, we only forward the signal if its not a SIGINT or if we are on Windows. And if there is child process, of course.

The output on Linux stays the same, but on Windows, we now get:

>python subprocs_3.py term
A started
B started
C started
B got a SIGBREAK
C got a SIGBREAK
A ended

It finally works on Linux, OS X and Windows! \o/

Conclusion

If you don’t have to (or don’t want to) support Windows, handling sub-processes and sending signals to them is not that hard. But I hope I could demonstrate that it’s—despite of all problems—possible to get the job done on Windows, too, without too much overhead.

An alternative to signals, if you are using some other kind of IPC anyways, might be using it to send a termination requests to your processes instead of sending a signal. There would be no need for the CREATE_NEW_PROCESS_GROUP flag then and Ctrl-C would work nicely on all platforms. You may take care though, that clients are not able to send a termination message to your server, but that depends on your use case.

In the end, I wish Windows was Posix-compliant. Would make life so much easier …

Bitbucket stopped rendering README.txt for Python projects [resolved]

2013-01-11T09:34:29+01:00

Until recently, Bitbucket nicely rendered reStructuredText (reST) formatted README.txt files for Python projects. This made totally sense, because PyPI requires the project’s description in reST and most people put it into README.txt files.

However, for some reason the Bitbucket guys removed that feature and it seems that they don’t intend to bring it back. Instead, you are urged to rename your files from *.txt to *.rst.

Maybe some feedback would help to bring that feature back …

Update: Bitbucket fixed that issue. :-)

SimPy 3 Preview

2012-12-12T15:15:22+01:00

SimPy is a process-based and event-driven simulation framework written in pure Python. It can also be used for multi-agent systems and other eventloop-based applications.

After several months of work and various iterations of SimPy’s new API, we can finally present a preview with the most important features working.

Here is a simple example:

>>> import simpy
>>>
>>> def clock(env):
...     while True:
...         print(env.now)
...         yield env.timeout(1)
...
>>> env = simpy.Environment()
>>> env.start(clock(env))
Process(clock)
>>> simpy.simulate(env, until=3)
0
1
2

You can find the full announcement at Google+ and on our mailing list.

The code is at Bitbucket, the documentation at Read the Docs.

Check Python site-packages for Updates

2012-04-14T11:50:50+02:00

A while ago, I found a nice little script called check_for_updates.py which uses PIP to check your installed Python packages for updates. However, it didn’t work under Python 3, so I ported it myself:

"""
Use pip to get a list of local packages to check against one or more package
indexes for updated versions.

"""
try:
    from cStringIO import StringIO
    import xmlrpclib
except ImportError:
    from io import StringIO
    import xmlrpc.client as xmlrpclib

from distutils.version import StrictVersion, LooseVersion
import sys

import pip


def get_local_packages():
    """
    Call pip's freeze -l

    returns a list of package_name, version tuples

    """
    sys.stdout = mystdout = StringIO()
    pip.main(['freeze', '-l'])
    sys.stdout = sys.__stdout__

    pkgs = mystdout.getvalue().split('\n')
    return [p.split('==') for p in pkgs]


def find_current_version(package, index_urls=None):
    """
    Using the XMLRPC method available for PyPI, get the most recent version
    of <package> from each of the index_urls and figure out which one (if any)
    is higher

    Returns a tuple of the index with the higher version and the version it has

    """
    if index_urls is None:
        index_urls = ['http://pypi.python.org/pypi']
    cur_version = '0'
    cur_index = ''
    for index_url in index_urls:
        pypi = xmlrpclib.ServerProxy(index_url, xmlrpclib.Transport())
        pypi_hits = pypi.package_releases(package)
        if len(pypi_hits) > 0:
            if compare_versions(pypi_hits[0], cur_version):
                cur_version = pypi_hits[0]
                cur_index = index_url

    return cur_index, cur_version


def compare_versions(version1, version2):
    """
    Compare 2 versions, starting with StrictVersion, and falling back on
    LooseVersion. Returns ``True`` if *version1* is greater than *version2*.

    """
    try:
        return StrictVersion(version1) > StrictVersion(version2)
    # in case of abnormal version number, fall back to LooseVersion
    except ValueError:
        return LooseVersion(version1) > LooseVersion(version2)


def output_line(pkg_name, new_version, old_version, index_url):
    """Output the line showing the formatted information."""
    msg = '%(bd)s%(pkg_name)s%(nm)s (%(new)s) via %(index)s. Currently %(old)s.'
    params = {
        'bd': BOLD,
        'nm': NORMAL,
        'pkg_name': pkg_name,
        'new': new_version,
        'old': old_version,
        'index': index_url,
    }
    print(msg % params)


NEWER = lambda x, y: compare_versions(str(x), y) == 1


if __name__ == '__main__':
    import curses
    curses.setupterm()
    CLEAR_SCREEN = curses.tigetstr('clear').decode('utf-8')
    BOLD = curses.tigetstr('bold').decode('utf-8')
    NORMAL = curses.tigetstr('sgr0').decode('utf-8')

    if len(sys.argv) > 1:
        indexes = sys.argv[1:]
    else:
        indexes = ['http://pypi.python.org/pypi']
    print(CLEAR_SCREEN + BOLD + 'Packages with newer versions:' + NORMAL)
    print('')

    for pkg in get_local_packages():
        # pip outputs a single 0 at the end of the list. Ignore it.
        if len(pkg) < 2:
            continue

        index, current_version = find_current_version(pkg[0], index_urls=indexes)
        if current_version and NEWER(str(current_version), pkg[1]):
            output_line(pkg[0], current_version, pkg[1], index)

Just execute it from anywhere using a Python interpreter whose packages you’d like to check (e.g., python3 check_for_updates.py or from within a virtualenv). You can then uses pip install -U <packagename> (or pip-3.2 for Python 3) to update the packages.

Update: The yolk package does the same via yolk -U. It doesn’t support Python 3, though. I wonder when PIP will get that functionality.

A Simple Web Bot with Requests and BeautifulSoup

2012-02-21T12:59:51+01:00

Today I helped a colleague debugging a web bot written in Java. Since I did’t really work with Java since a few years, I thought it would be easier for me to reproduce (and solve) the problem with Requests and BeautifulSoup. (I’ve actually been looking for an opportunity to try Requests out for a while, since I’ve heard so much good about it.)

And what can I say—I was blown away by how easy it was to implement a simple web bot that filled out a form and grabbed a huge table of data for me.

I cannot show you exactly that web bot, but here’s how you could search my website for “Python” and get the headings of the resulting posts:

from BeautifulSoup import BeautifulSoup
import requests

url = 'http://stefan.sofa-rockers.org/search/?q=%(q)s'
payload = {
    'q': 'Python',
}
r = requests.get(url % payload)

soup = BeautifulSoup(r.text)
titles = [h2.text for h2 in soup.findAll('h2', attrs={'class': 'post_title'})]

for t in titles:
    print(t)

Designing and Testing PyZMQ Applications – Part 3

2012-02-15T18:20:22+01:00

The third and last part of this series is again just about testing. While the previous article focused on unit testing, this one will be about testing complete PyZMQ processes. This even involves some magic!

Once you’ve made sure that your message dispatching and application logic works fine, you can actually start sending real messages to your process and checking real replies. This can be done for single processes—I call this process testing—and for your complete application (system testing).

When you test a single process, you create sockets that mimic all processes the tested process communicates with. When you do a system test, you only mimic a client or just invoke your program from the command line and check its output (e.g., what it prints to stdtout and stderr or results written to a database).

I’ll start with process testing, which is a bit more generalizable than system testing.

Process Testing

The biggest problem I ran into when I started testing processes was that I often made blocking calls to recv methods and these halted my tests and gave me no output about what actually went wrong. Though you can make them non- blocking by passing zmq.NOBLOCK as an extra argument, this doesn’t solve your problems. You will now need a very precise timing and many time.sleep(x) calls, because recv will instantly raise an error if there is nothing to be received.

My solution for this was to wrap PyZMQ sockets and add a timeout to its send and recv methods. The following wrapper will try to receive something for one second and raise an exception if that failed. There’s also a simple wrapper for methods like connect or bind, but it’s really not that interesting, so I’ll omit it here.

# test/support.py

def get_wrapped_fwd(func):
    """
    Returns a wrapper, that tries to call *func* multiple time in non-blocking
    mode before rasing an :class:`zmq.ZMQError`.

    """
    def forwarder(*args, **kwargs):
        # 100 tries * 0.01 second == 1 second
        for i in range(100):
            try:
                rep = func(*args, flags=zmq.NOBLOCK, **kwargs)
                return rep

            except zmq.ZMQError:
                time.sleep(0.01)

        # We should not get here, so raise an error.
        msg = 'Could not %s message.' % func.__name__[:4]
        raise zmq.ZMQError(msg)

    return forwarder

This wrapper is now used to create a TestSocket class with the desired behavior:

# test/support.py

class TestSocket(object):
    """
    Wraps ZMQ :class:`~zmq.core.socket.Socket`. All *recv* and *send* methods
    will be called multiple times in non-blocking mode before a
    :class:`zmq.ZMQError` is raised.

    """
    def __init__(self, context, sock_type):
        self._context = context

        sock = context.socket(sock_type)
        self._sock = sock

        forwards = [  # These methods can simply be forwarded
            sock.bind,
            sock.bind_to_random_port,
            sock.connect,
            sock.close,
            sock.setsockopt,
        ]
        wrapped_fwd = [  # These methods are wrapped with a for loop
            sock.recv,
            sock.recv_json,
            sock.recv_multipart,
            sock.recv_unicode,
            sock.send,
            sock.send_json,
            sock.send_multipart,
            sock.send_unicode,
        ]

        for func in forwards:
            setattr(self, func.__name__, get_forwarder(func))

        for func in wrapped_fwd:
            setattr(self, func.__name__, get_wrapped_fwd(func))

In order to reuse the same ports for all test methods, you need to cleanly close all sockets after each test. To handle method level setup/teardown in pytest, you need to implement a setup_method and a teardown_method. In the setup method, you create one or more TestSocket instances that mimic other processes and you also start the process to be tested:

# test/process/test_pongproc.py
import pytest
import zmq

from test.support import ProcessTest, make_sock
import pongproc


host = '127.0.0.1'
port = 5678


class TestProngProc(ProcessTest):
    """Communication test for the Platform Manager process."""

    def setup_method(self, method):
        """
        Creates and starts a PongProc process and sets up sockets to
        communicate with it.

        """
        self.context = zmq.Context()

        # make_sock creates and connects a TestSocket that we will use to
        # mimic the Ping process
        self.req_sock = make_sock(self.context, zmq.REQ,
                                  connect=(host, port))

        self.pp = pongproc.PongProc((host, port))
        self.pp.start()

    def teardown_method(self, method):
        """
        Sends a kill message to the pp and waits for the process to terminate.

        """
        # Send a stop message to the prong process and wait until it joins
        self.req_sock.send_multipart([b'["plzdiekthxbye", null]'])
        self.pp.join()

        self.req_sock.close()

You may have noticed that our test class inherits ProcessTests. This class and some helpers in a conftest.py allow us to use some magic that improves the readability of the actual test:

# test/process/test_pongproc.py

    def test_ping(self):
        """Tests a ping-pong sequence."""
        yield ('send', self.req_sock, [], ['ping', 1])

        reply = yield ('recv', self.req_sock)
        assert reply == [['pong', 1]]

You can just yield send or recv events from your test case! When you yield a send, the test machinery tries to send a message via the specified socket. When you yield a receive, ProcessTest tries to receive something from the socket and sends its result back to your test function, so that you can easily compare the reply with the expected result.

The example above is roughly equivalent to the following code:

self.req_sock.send_multipart([] + [json.dumps(['ping', 1])])

reply = self.req_sock.recv_multipart()
reply[-1] = json.loads[reply[-1]]
assert reply == [['pong', 1]]

So how does this work? By default, if pytests finds a test function that is a generator, it assumes that it generates further test functions. Hence, our first step is to override this behavior. We can do this in a conftest.py file in the test/process/ directory by implementing a pytest_pycollect_makeitem function. In this case, we collect generator functions like normal functions:

# test/process/conftest.py
from inspect import isfunction, isgeneratorfunction


def pytest_pycollect_makeitem(collector, name, obj):
    """
    Collects all instance methods that are generators and returns them as
    normal function items.

    """
    if collector.funcnamefilter(name) and hasattr(obj, '__call__'):
        if isfunction(obj) or isgeneratorfunction(obj):
            return collector._genfunctions(name, obj)

Now, we need to tell pytest how to run a test on the collected generator functions. This can be done by implementing pytest_runtest_call. If the object we are going to test (item.obj) is a generator function, we call the run method of the object’s instance (item.obj.__self__.run) and pass the generator function to it. If the test item contains a normal function, we run the default test.

# test/process/conftest.py

def pytest_runtest_call(item):
    """
    Passes the test generator (``item.obj``) to the ``run()`` method of the
    generator's instance. This method should be inherited from
    :class:`test.support.ProcessTest`.

    """
    if isgeneratorfunction(item.obj):
        item.obj.__self__.run(item.obj)
    else:  # Normal test execution for normal instance methods
        item.runtest()

But wait—we didn’t implement a run method in our test case! So it must be inherited from ProcessTest. Let’s take a look at it:

# test/support.py
class ProcessTest(object):
    """
    Base class for process tests. It offers basic actions for sending and
    receiving messages and implements the *run* methods that handles the
    actual test generators.

    """

    def run(self, testfunc):
        """
        Iterates over the *testfunc* generator and executes all actions it
        yields. Results will be sent back into the generator.

        :param testfunc: A generator function that yields tuples containing
                an action keyword, which should be a function of this or
                the inheriting class (like ``send`` or ``recv``) and additional
                parameters that will be passed to that function, e.g.:
                ``('send', socket_obj, ['header'], 'body')``
        :type testfunc:  generatorfunction

        """
        item_gen = testfunc()
        item = next(item_gen)

        def throw_err(skip_levels=0):
            """
            Throws the last error to *item_gen* and skips *skip_levels* in
            the traceback to point to the line that yielded the last event.

            """
            etype, evalue, tb = sys.exc_info()
            for i in range(skip_levels):
                tb = tb.tb_next
            item_gen.throw(etype, evalue, tb)

        try:
            while True:
                try:
                    # Call the event handler and pass the args,
                    # e.g., self.send(socket_obj, header, body)
                    ret = getattr(self, item[0])(*item[1:])

                    # Send the results back to the test and get the next item
                    item = item_gen.send(ret)

                except zmq.ZMQError:
                    throw_err(3)  # PyZMQ could not send/recv
                except AssertionError:
                    throw_err(1)  # Error in the test
        except StopIteration:
            pass

The run method simply iterates over all events our testfunc generates and calls a method with the name of the event (e.g., send or recv). Their return value is sent back into the generator. If an error occurs, the exception’s traceback is modified to point to the line of code that yielded the according event and not to the run method itself.

The methods send and recv roughly do the same as the snippet I showed you above:

# test/support.py

    def send(self, socket, header, body, extra_data=[]):
        """
        JSON-encodes *body*, concatenates it with *header*, appends
        *extra_data* and sends it as multipart message over *socket*.

        *header* and *extra_data* should be lists containg byte objects or
        objects implementing the buffer interface (like NumPy arrays).

        """
        socket.send_multipart(header + [json.dumps(body)] + extra_data)

    def recv(self, socket, json_load_index=-1):
        """
        Receives and returns a multipart message from *socket* and tries to
        JSON-decode the item at position *json_load_index* (defaults to ``-1``;
        the last element in the list). The original byte string will be
        replaced by the loaded object. Set *json_load_index* to ``None`` to get
        the original, unchanged message.

        """
        msg = socket.recv_multipart()
        if json_load_index is not None:
            msg[json_load_index] = json.loads(msg[json_load_index])
        return msg

You can even add your own event handler to your test class. I used this, for example, to add a log event that checks if a PyZMQ log handler sent the expected log messages:

def log(self, substr=''):
    """
    Receives a message and asserts, that it is a log message and that
    *substr* is in that message.

    Usage:
        yield ('log', 'Ai iz in ur log mesage')

    """
    msg = self.log_sock.recv_json()
    assert msg[0] == 'log_message'
    assert substr in msg[1]

What if your process starts further subprocesses?

In some cases, the process you are about to test starts additional subprocesses that you don’t want to test. Even worse, these processes might communicate via sockets bound to random ports. And EVEN WORSE, the process you are testing might depend on excepting a KeyboardInterrupt to send stop messages to child processes or to clean something up!

The last problem is quite easy to solve: You just a send a SIGINT to your process from the test:

import os, signal

def teardown_method(self, method):
    os.kill(self.my_proc.pid, signal.SIGINT)
    self.my_proc.join()

    # Now you can close the test sockets

If you don’t want to start a certain subprocess, you can just mock it. Imagine, you have two processes a.A and b.B, where A starts B, then you just mock B before starting A:

with mock.patch('b.B'):
    self.a = A()
    self.a.start()

Imagine now, that A binds a socket to a random port and uses that socket to communicate with B. If you want to mock B in your tests, you need that port number in order to connect to it and send messages to A.

But how can you get that number? When A creates B, it already runs in its own process, so a simple attribute access won’t work. Setting a random seed would only work if you did that directly in A when it’s already running. But doing that just for the tests is not such a good idea. It also may not work reliably on all systems and Python versions.

However, A must pass the socket number to B, so that B can connect to A. Thus, we can create a mock for B that will send us its port number via a queue <http://docs.python.org/py3k/library/multiprocessing#exchanging-objects- between-processes>:

class ProcMock(mock.Mock):
    """
    This mock returns itself when called, so it acts like both, the
    process’ class and instance object.

    """
    def __init__(self):
        super().__init__()
        self.queue = multiprocessing.Queue()

    def __call__(self, port):
        """Will be called when A instantiates B and passes its port number."""
        self.queue.put(port)
        return self

    def start(self):
        return  # Just make sure the methods exists and returns nothing

    def join(self):
        return  # Just make sure the methods exists and returns nothing


class TestA(ProcessTest):

    def setup_method(self):

        b_mock = ProcMock()
        with mock.patch('b.B', new=b_mock):
            self.a = A()
            self.a.start()

        # Get the port A is listening on
        port = b_mock.queue.get()

        # ...

As you’ve seen, process testing is really not as simple as unit testing. But I always found bugs with it that my unit tests coudn’t detect. If you cover all communication sequences for a process in a process test, you can be pretty sure, that it will also work flawlessly in the final application.

System Testing

If your application consists of more than one process, you still need to test whether all processes work nicely together or not. This is something you cannot simulate reliably with a process tests, as much as unit tests can’t replace the process test.

Writing a good system test is very application-specific and can, depending on the complexity of your application, be very hard or very easy. Fortunately, the latter is the case for our ping-pong app. We just start it and copy its output to a file. If the output is not what we expected, we modify the file accordingly. In our test, we can now simply invoke our programm again, capture its output and compare it to the contents of the file we created before:

# test/system/test_pongproc.py
import os.path
import subprocess

import pytest


def test_pongproc():
    filename = os.path.join('test', 'data', 'pongproc.out')
    expected = open(filename).read()

    output = subprocess.check_output(['python', 'pongproc.py'],
                                     universal_newlines=True)

    assert output == expected

If your application was a server, another way of doing the system test would be to emulate a client that speaks with it. Your system test would then be very similar to your process tests, except that you only mimic the client and not all processes your main process communicates with.

Other applications (like, for instance, simulations) might create a database containing collected data. Here, you might check if these results match your expectations.

Of course you can also combine these possibilities or do something completely different …

“My Test Are now Running sooo Slow!”

System and process tests often run much slower than simple unit tests, so you may want to skip them most of the time. Pytest allows you to mark a test with a given name. You can then (de)select tests based on their mark when you invoke pytest.

To mark a module e.g. as process test, just put a line pytestmark = pytest.mark.process somewhere in it. Likewise, you can add a pytestmark = pytest.mark.system to mark a module as system test.

You can now deselect process and system tests:

$ py.test -m "not (process or system)"

You can put this into a pytest.ini as a default setting. To override this again, use 1 or True as selection expression:

$ py.test -m 1

Summary

Process and system testing were the last two topics I wanted to cover in this series. Compared to simple unit tests, they require a bit more effort. I think they are definitely worth the extra work since they give you a lot more confidence that your program actually works, because they are much more realistic than unit tests can be.

In the end, these articles became much longer and more elaborate then I originally planned them to be. However, I hope they provided a good overview about how to design and test applications with PyZMQ and I hope that you now run into much less problems than I did when I first started working with PyZMQ.

Designing and Testing PyZMQ Applications – Part 2

2012-02-07T18:54:47+01:00

This is the second part of the series Designing and Testing PyZMQ Applications. In the first part, I wrote about designing a PyZMQ application, so this time it’s all about (unit) testing (remember, if it’s not tested, it’s broken). I also updated the repository for this article with the new code examples.

My favorite testing tools are pytest by Holger Krekel and Mock by Michael Ford. Pytest is particularly awesome because of its re-evaluation of assert statements. If your test contains an assert spam == 'eggs' and the assert fails, pytest re-evaluates it and prints the value of spam. Really helpful and you don’t need any boilerplate code for that. Mock is really nice for mocking external dependencies and asserting that your code called them in the correct way.

If you cloned the repository for this article, just run py.test from its root directory:

$ pip install pytest mock
...
Successfully installed pytest mock
Cleaning up...
$ py.test
=================== test session starts ====================
platform darwin -- Python 3.2.2 -- pytest-2.2.3
collected 11 items

example_app/test/test_base.py ....
example_app/test/test_pongproc.py .......

================ 11 passed in 0.12 seconds =================

Unit Testing

The probability that PyZMQ works correctly is very high. The probability that your code will call a PyZMQ function in such a way that it blocks forever and halts your test runner is also very high. Therefore, it’s a good idea to mock everything PyZMQ-related for your unit tests. And since your application logic might also not be implemented when you start testing your process, you should mock that, too.

What you’ll actually end up testing is the following:

Does your message handler call your application logic in the right way given a certain input message?
Does your message handler create and send the correct reply based on the return value of your application logic?

ZmqProcess

Let’s start with ZmqProcess again. After all, everything else depends on it. Testing its setup method is easy. We just check that it creates a context and a loop:

# example_app/test/test_zmqproc.py
from zmq.eventloop import ioloop
import mock
import pytest
import zmq

import zmqproc


class TestZmqProcess(object):
    """Tests for :class:`base.ZmqProcess`."""

    def test_setup(self):
        zp = base.ZmqProcess()
        zp.setup()

        assert isinstance(zp.context, zmq.Context)
        assert isinstance(zp.loop, ioloop.IOLoop)

Testing stream is more complicated. We need to test if it can handle various address formats, if it creates or binds correctly and if it performs a default subscription for SUB sockets.

Pytest 2.2 introduced a parametrize decorator, that helps calling a test multiple times with varying inputs. You just define one or more arguments for your test function and a list of values for these arguments. For test_stream, I only need a kwargs parameter containing the parameters for the stream call:

# example_app/test/test_zmqproc.py

    @pytest.mark.parametrize('kwargs', [
        dict(sock_type=23, addr='127.0.0.1:1234', bind=True,
              callback=mock.Mock()),
        dict(sock_type=23, addr='127.0.0.1', bind=True,
              callback=mock.Mock()),
        dict(sock_type=zmq.SUB, addr=('localhost', 1234), bind=False,
              callback=mock.Mock(), subscribe=b'ohai'),
    ])
    def test_stream(self, kwargs):

The next step is to create an instance of ZmqProcess and patch some of its attributes. We also need to set a defined return value for the socket’s bind_to_random_port method:

# example_app/test/test_zmqproc.py

        zp = base.ZmqProcess()

        # Patch the ZmqProcess instance
        zp.context = mock.Mock(spec_set=zmq.Context)
        zp.loop = mock.Mock(spec_set=ioloop.IOLoop)
        sock_mock = zp.context.socket.return_value
        sock_mock.bind_to_random_port.return_value = 42

For the actual test, we also need to patch ZMQStream. Although mock.patch could work as a function decorator, we need to use it as context processor if we also uses pytest funcargs (e.g., via the parametrize decorator—I don’t know if it’s even possible to uses both, mock.patch as decorator and pytest funcargs in one test).

# example_app/test/test_zmqproc.py

        # Patch ZMQStream and start testing
        with mock.patch('zmq.eventloop.zmqstream.ZMQStream') as zmqstream_mock:
            stream, port = zp.stream(**kwargs)

Finally, we can check the return values of our stream method and it made the correct calls to create the stream:

# example_app/test/test_zmqproc.py

            # Assert that the return values are correct
            assert stream is zmqstream_mock.return_value
            if isinstance(kwargs['addr'], tuple):
                assert port == kwargs['addr'][1]
            elif ':' in kwargs['addr']:
                assert port == int(kwargs['addr'][-4:])
            else:
                assert port == sock_mock.bind_to_random_port.return_value

            # Check that the socket was crated correctly
            assert zp.context.socket.call_args == ((kwargs['sock_type'],), {})
            if kwargs['bind'] and ':' in kwargs['addr']:
                assert sock_mock.bind.call_args == (
                        ('tcp://%s' % kwargs['addr'],), {})
            elif kwargs['bind']:
                assert sock_mock.bind_to_random_port.call_args == (
                        ('tcp://%s' % kwargs['addr'],), {})
            else:
                assert sock_mock.connect.call_args == (
                        ('tcp://%s:%s' % kwargs['addr'],), {})

            # Check creation of the stream
            assert zmqstream_mock.call_args == ((sock_mock, zp.loop), {})
            assert zmqstream_mock.return_value.on_recv.call_args == (
                    (kwargs['callback'],), {})

            # Check default subscribtion
            if 'subscribe' in kwargs:
                assert sock_mock.setsockopt.call_args == (
                        (zmq.SUBSCRIBE, kwargs['subscribe']), {})

Note: You may have noticed that I use assert my_mock.call_args == ... rather than my_mock.assert_called_with(...). The reason for that is simply, that assert statements are highlighted but ordinary function calls are not. This makes it easier for me to find all assertions in a test.

MessageHandler

The MessageHandler base class has only one methd, __call__, but I split the test for it into two methods—one that tests the JSON-loading functionality and one that checks if the correct handler method is called:

# example_app/test/test_base.py

class TestMessageHandler(object):
    """Tests for :class:`base.TestMessageHandler`."""

    @pytest.mark.parametrize(('idx', 'msg'), [
        (-1, [23, b'["test", null]']),
        (1, [23, b'["test", "spam"]', 42]),
        (TypeError, [23, 42]),
        (ValueError, [23, b'["test"]23spam']),
    ])
    def test_call_json_load(self, idx, msg):
        handler = mock.Mock()
        mh = base.MessageHandler(idx if isinstance(idx, int) else -1)
        mh.test = handler

        if isinstance(idx, int):
            mh(msg)
            assert handler.call_count == 1
        else:
            pytest.raises(idx, mh, msg)

    @pytest.mark.parametrize(('ok', 'msg'), [
        (True, [23, b'["test", "spam"]', 42]),
        (AttributeError, [23, b'["_test", "spam"]', 42]),
        (TypeError, [23, b'["spam", "spam"]', 42]),
        (AttributeError, [23, b'["eggs", "spam"]', 42]),
    ])
    def test_call_get_handler(self, ok, msg):
        handler = mock.Mock()
        mh = base.MessageHandler(1)
        mh.test = handler
        mh.spam = 'spam'

        if ok is True:
            mh(msg)
            assert handler.call_args == (
                    (msg[0], 'spam', msg[2]), {})
        else:
            pytest.raises(ok, mh, msg)

PongProc

Testing the PongProc is not much different from testing its base class. pytest_funcarg__pp will instantiate a PongProc instance for each test that has a pp argument. The tests for setup, run and stop are easy to do. We create a few mocks and then ask them if the tested function called them correctly:

# example_app/test/test_pongproc.py
from zmq.utils import jsonapi as json
import mock, pytest, zmq

import pongproc

host, port = '127.0.0.1', 5678

def pytest_funcarg__pp(request):
    """Creates a PongProc instance."""
    return pongproc.PongProc((host, port))


class TestPongProc(object):
    """Tests :class:`pongproc.PongProc`."""

    def test_setup(self, pp):
        def make_stream(*args, **kwargs):
            stream = mock.Mock()
            stream.type = args[0]
            return stream, mock.Mock()
        pp.stream = mock.Mock(side_effect=make_stream)

        with mock.patch('base.ZmqProcess.setup') as setup_mock:
            pp.setup()
            assert setup_mock.call_count == 1

        assert pp.stream.call_args_list == [
            ((zmq.REP, (host, port)), dict(bind=True)),
        ]
        assert pp.rep_stream.type == zmq.REP

        # Test if the message handler was configured correctly
        rsh = pp.rep_stream.on_recv.call_args[0][0]  # Get the msg handler
        assert rsh._rep_stream == pp.rep_stream
        assert rsh._stop == pp.stop

    def test_run(self, pp):
        pp.setup = mock.Mock()
        pp.loop = mock.Mock()

        pp.run()

        assert pp.setup.call_count == 1
        assert pp.loop.start.call_count == 1

    def test_stop(self, pp):
        pp.loop = mock.Mock()
        pp.stop()
        assert pp.loop.stop.call_count == 1

RepStreamHandler

Testing the actual message handler requires some mocks, but is apart from that straight forward. A funcarg method creates an instance of the message handler for each test case which we feed with a message. We than check if the application logic was called correctly and/or if a correct reply is sent:

# example_app/test/test_pongproc.py

def pytest_funcarg__rsh(request):
    """Creates a RepStreamHandler instance."""
    return pongproc.RepStreamHandler(
            rep_stream=mock.Mock(),
            stop=mock.Mock(),
            ping_handler=mock.Mock(spec_set=pongproc.PingHandler()))


class TestRepStreamHandler(object):
    def test_ping(self, rsh):
        msg = ['ping', 1]
        retval = 'spam'
        rsh._ping_handler = mock.Mock(spec_set=pongproc.PingHandler)
        rsh._ping_handler.make_pong.return_value = retval

        rsh([json.dumps(msg)])

        assert rsh._ping_handler.make_pong.call_args == ((msg[1],), {})
        assert rsh._rep_stream.send_json.call_args == ((retval,), {})

    def test_plzdiekthybye(self, rsh):
        rsh([b'["plzdiekthxbye", null]'])
        assert rsh._stop.call_count == 1

PingHandler

When we are done with all that network stuff, we can finally test the application logic. Easy-peasy in our case:

# example_app/test/test_pongproc.py

def pytest_funcarg__ph(request):
    """Creates a PingHandler instance."""
    return pongproc.PingHandler()

class TestPingHandler(object):
    def test_make_pong(self, ph):
        ping_num = 23
        ret = ph.make_pong(ping_num)
        assert ret == ['pong', ping_num]

Summary

Thanks to the Mock library, unit testing PyZMQ apps is really not that hard and not much different from normal unit testing. However, what we know now is only, that our process should work in theory. We haven’t yet started it and sent real messages to it.

The next and final part of this series will show you how you can automate testing complete processes. Until then, you should get your test coverage up to 100% to protect yourself from nasty surprises when you start with process testing.

Designing and Testing PyZMQ Applications – Part 1

2012-02-01T08:05:33+01:00

ZeroMQ (or ØMQ or ZMQ) is an intelligent messaging framework and described as “sockets on steroids”. That is, they look like normal TCP sockets but actually work as you’d expect sockets to work. PyZMQ adds even more convenience to them, which makes it a really a good choice if you want to implement a distributed application. Another big plus for ØMQ is that you can integrate sub-systems written in C, Java or any other language ØMQ supports (which are a lot).

If you’ve never heard of ØMQ before, I recommend to read ZeroMQ an Introduction by Nicholas Piël, before you go on with this article.

The ØMQ Guide and PyZMQ’s documentation are really good, so you can easily get started. However, when we began to implement a larger application with it (a distributed simulation framework), several questions arose which were not covered by the documentation:

What’s the best way do design our application?
How can we keep it readable, flexible and maintainable?
How do we test it?

I didn’t find something like a best practice article that answered my questions. So in this series of articles, I’m going to talk about what I’ve learned during the last months. I’m not a PyZMQ expert (yet ;-)), but what I’ve done so far works quite well and I never had more tests in a project than I do have now.

You’ll find the source for the examples at bitbucket. They are written in Python 3.2 and tested under Mac OS X Lion, Ubuntu 11.10 and Windows 7, 64 bit in each case. If you have any suggestions or improvements, please fork me or just leave a comment.

In this first article, I’m going to talk a bit about how you could generally design your application to be flexible, maintainable and testable. The second part will be about unit testing and the finally, I’ll cover process and system testing.

Comparison of Different Approaches

There are basically three possible ways to implement a PyZMQ application. One, that’s easy, but limited in practical use, one that’s more flexible, but not really pythonic and one, that needs a bit more setup, but is flexible and pythonic.

All three examples feature a simple ping process and a pong process with varying complexity. I use multiprocessing to run the pong process, because that’s what you should usually do in real PyZMQ applications (you don’t want to use threads and if both processes are running on the same machine, there’s no need to invoke both of them separately).

All of the examples will have the following output:

(zmq)$ python blocking_recv.py
Pong got request: ping 0
Ping got reply: pong 0
...
Pong got request: ping 4
Ping got reply: pong 4

Let’s start with the easy one first. You just use on of the socket’s recv methods in a loop:

# blocking_recv.py
import multiprocessing
import zmq


addr = 'tcp://127.0.0.1:5678'


def ping():
    """Sends ping requests and waits for replies."""
    context = zmq.Context()
    sock = context.socket(zmq.REQ)
    sock.bind(addr)

    for i in range(5):
        sock.send_unicode('ping %s' % i)
        rep = sock.recv_unicode()  # This blocks until we get something
        print('Ping got reply:', rep)


def pong():
    """Waits for ping requests and replies with a pong."""
    context = zmq.Context()
    sock = context.socket(zmq.REP)
    sock.connect(addr)

    for i in range(5):
        req = sock.recv_unicode()  # This also blocks
        print('Pong got request:', req)
        sock.send_unicode('pong %s' % i)


if __name__ == '__main__':
    pong_proc = multiprocessing.Process(target=pong)
    pong_proc.start()

    ping()

    pong_proc.join()

So this is very easy and no that much code. The problem with this is, that it only works well if your process only uses one socket. Unfortunately, in larger applications that is rather rarely the case.

A way to handle multiple sockets per process is polling. In addition to your context and socket(s), you need a poller. You also have to tell it which events on which socket you are going to poll:

# polling.py
def pong():
    """Waits for ping requests and replies with a pong."""
    context = zmq.Context()
    sock = context.socket(zmq.REP)
    sock.bind(addr)

    # Create a poller and register the events we want to poll
    poller = zmq.Poller()
    poller.register(sock, zmq.POLLIN|zmq.POLLOUT)

    for i in range(10):
        # Get all sockets that can do something
        socks = dict(poller.poll())

        # Check if we can receive something
        if sock in socks and socks[sock] == zmq.POLLIN:
            req = sock.recv_unicode()
            print('Pong got request:', req)

        # Check if we cann send something
        if sock in socks and socks[sock] == zmq.POLLOUT:
            sock.send_unicode('pong %s' % (i // 2))

    poller.unregister(sock)

You see, that our pong function got pretty ugly. You need 10 iterations to do five ping-pongs, because in each iteration you can either send or reply. And each socket you add to your process adds two more if-statements. You could improve that design if you created a base class wrapping the polling loop and just register sockets and callbacks in an inheriting class.

That brings us to our final example. PyZMQ comes with with an adapted Tornado eventloop that handles the polling and works with ZMQStreams, that wrap sockets and add some functionality:

# eventloop.py
from zmq.eventloop import ioloop, zmqstream


class Pong(multiprocessing.Process):
    """Waits for ping requests and replies with a pong."""
    def __init__(self):
        super().__init__()
        self.loop = None
        self.stream = None
        self.i = 0

    def run(self):
        """
        Initializes the event loop, creates the sockets/streams and
        starts the (blocking) loop.

        """
        context = zmq.Context()
        self.loop = ioloop.IOLoop.instance()  # This is the event loop

        sock = context.socket(zmq.REP)
        sock.bind(addr)
        # We need to create a stream from our socket and
        # register a callback for recv events.
        self.stream = zmqstream.ZMQStream(sock, self.loop)
        self.stream.on_recv(self.handle_ping)

        # Start the loop. It runs until we stop it.
        self.loop.start()

    def handle_ping(self, msg):
        """Handles ping requests and sends back a pong."""
        # req is a list of byte objects
        req = msg[0].decode()
        print('Pong got request:', req)
        self.stream.send_unicode('pong %s' % self.i)

        # We’ll stop the loop after 5 pings
        self.i += 1
        if self.i == 5:
            self.stream.flush()
            self.loop.stop()

This even adds more boilerplate code, but it will pay of if you use more sockets and most of that stuff in run() can be put into a base class. Another drawback is, that the IOLoop only uses recv_multipart(). So you always get a lists of byte strings which you have to decode or deserialize on your own. However, you can use all the send methods socket offers (like send_unicode() or send_json()). You can also stop the loop from within a message handler.

In the next sections, I’ll discuss how you could implement a PyZMQ process that uses the event loop.

Communication Design

Before you start to implement anything, you should think about what kind of processes you need in your application and which messages they exchange. You should also decide what kind of message format and serialization you want to use.

PyZMQ has built-in support for Unicode (send sends plain C strings which map to Python byte objects, so there’s a separate method to send Unicode strings), JSON and Pickle.

JSON is nice, because it’s fast and lets you integrate processes written in other languages into you application. It’s also a bit safer, because you cannot receive arbitrary objects as with pickle. The most straightforward syntax for JSON messages is to let them be triples [msg_type, args, kwargs], where msg_type maps to a method name and args and kwargs get passed as positional and keyword arguments.

I strongly recommend you to document each chain of messages your application sends to perform a certain task. I do this with fancy PowerPoint graphics and with even fancier ASCII art in Sphinx. Here is how I would document our ping-pong:

Sending pings
-------------

* If the ping process sends a *ping*, the pong processes responds with a
  *pong*.
* The number of pings (and pongs) is counted. The current ping count is
  sent with each message.

::

    PingProc      PongProc
     [REQ] ---1--> [REP]
           <--2---


    1 IN : ['ping, count']
    1 OUT: ['ping, count']

    2 IN : ['pong, count']
    2 OUT: ['pong, count']

First, I write some bullet points that explain how the processes behave and why they behave this way. This is followed by some kind of sequence diagram that shows when which process sents which message using which socket type. Finally, I write down how the messages are looking. # IN is what you would pass to send_multipart and # OUT is, what is received on the other side by recv_multipart. If one of the participating sockets is a ROUTER or DEALER, IN and OUT will differ (though that’s not the case in this example). Everything in single quotation marks (') represents a JSON serialized list.

If our pong process used a ROUTER socket instead of the REP socket, it would look like this:

1 IN : ['ping, count']
1 OUT: [ping_uuid, '', 'ping, count']

2 IN : [ping_uuid, '', 'pong, count']
2 OUT: ['pong, count']

This seems like a lot of tedious work, but trust me, it really helps a lot when you need to change something a few weeks later!

Application Design

In the examples above, the Pong process was responsible for setting everything up, for receiving/sending messages and for the actual application logic (counting incoming pings and creating a pong).

Obviously, this is not a very good design (at least if your application gets more complex than our little ping-pong example). What we can do about this is to put most of that nasty setup stuff into a base class which all your processes can inherit from, separate message handling and (de)serialization from it and finally put all the actual application logic into a separate (PyZMQ-independent) class. This will result in a three-level architecture:

The lowest tier will contain the entry point of the process, set-up everything and start the event loop. A common base class provides utilities for creating sockets/streams and setting everything up.
The second level is message handling and (de) serialization. A base class performs the (de)serialization and error handling. A message handler inherits this class and implements a method for each message type that should be handled.
The third level will be the application logic and completely PyZMQ-agnostic.

Base classes should be defined for the first two tiers two reduce redundant code in multiple processes or message handlers. The following figure shows the five classes our process is going to consist of:

The refactored PongProc now consists of three layers. The main class PongProc inherits ZMQProcess. Every stream gets a MessageHandler. In our example it’s just RepStreamHandler. Finally, you can have one ore more classes containing the (PyZMQ-agnostic) application logic. In our example, it’s called PingHandler, because it handles incoming pings.

ZmqPocess – The Base Class for all Processes

The base class basically implements two things:

a setup method that creates a context an a loop
a stream factory method for streams with a on_recv callback. It creates a socket and can connect/bind it to a given address or bind it to a random port (that’s why it returns the port number in addition to the stream itself).

It also inherits multiprocessing.Process so that it is easier to spawn it as sub-process. Of course, you can also just call its run() method from you main().

# example_app/base.py
import multiprocessing

from zmq.eventloop import ioloop, zmqstream
import zmq


class ZmqProcess(multiprocessing.Process):
    """
    This is the base for all processes and offers utility functions
    for setup and creating new streams.

    """
    def __init__(self):
        super().__init__()

        self.context = None
        """The ØMQ :class:`~zmq.Context` instance."""

        self.loop = None
        """PyZMQ's event loop (:class:`~zmq.eventloop.ioloop.IOLoop`)."""

    def setup(self):
        """
        Creates a :attr:`context` and an event :attr:`loop` for the process.

        """
        self.context = zmq.Context()
        self.loop = ioloop.IOLoop.instance()

    def stream(self, sock_type, addr, bind, callback=None, subscribe=b''):
        """
        Creates a :class:`~zmq.eventloop.zmqstream.ZMQStream`.

        :param sock_type: The ØMQ socket type (e.g. ``zmq.REQ``)
        :param addr: Address to bind or connect to formatted as *host:port*,
                *(host, port)* or *host* (bind to random port).
                If *bind* is ``True``, *host* may be:

                - the wild-card ``*``, meaning all available interfaces,
                - the primary IPv4 address assigned to the interface, in its
                numeric representation or
                - the interface name as defined by the operating system.

                If *bind* is ``False``, *host* may be:

                - the DNS name of the peer or
                - the IPv4 address of the peer, in its numeric representation.

                If *addr* is just a host name without a port and *bind* is
                ``True``, the socket will be bound to a random port.
        :param bind: Binds to *addr* if ``True`` or tries to connect to it
                otherwise.
        :param callback: A callback for
                :meth:`~zmq.eventloop.zmqstream.ZMQStream.on_recv`, optional
        :param subscribe: Subscription pattern for *SUB* sockets, optional,
                defaults to ``b''``.
        :returns: A tuple containg the stream and the port number.

        """
        sock = self.context.socket(sock_type)

        # addr may be 'host:port' or ('host', port)
        if isinstance(addr, str):
            addr = addr.split(':')
        host, port = addr if len(addr) == 2 else (addr[0], None)

        # Bind/connect the socket
        if bind:
            if port:
                sock.bind('tcp://%s:%s' % (host, port))
            else:
                port = sock.bind_to_random_port('tcp://%s' % host)
        else:
            sock.connect('tcp://%s:%s' % (host, port))

        # Add a default subscription for SUB sockets
        if sock_type == zmq.SUB:
            sock.setsockopt(zmq.SUBSCRIBE, subscribe)

        # Create the stream and add the callback
        stream = zmqstream.ZMQStream(sock, self.loop)
        if callback:
            stream.on_recv(callback)

        return stream, int(port)

PongProc – The Actual Process

The PongProc inherits ZmqProcess and is the main class for our process. It creates the streams, starts the event loop and dispatches all messages to the appropriate handlers:

# example_app/pongproc.py
import zmq

import base


host = '127.0.0.1'
port = 5678


class PongProc(base.ZmqProcess):
    """
    Main processes for the Ponger. It handles ping requests and sends back
    a pong.

    """
    def __init__(self, bind_addr):
        super().__init__()

        self.bind_addr = bind_addr
        self.rep_stream = None
        self.ping_handler = PingHandler()

    def setup(self):
        """Sets up PyZMQ and creates all streams."""
        super().setup()

        # Create the stream and add the message handler
        self.rep_stream, _ = self.stream(zmq.REP, self.bind_addr, bind=True)
        self.rep_stream.on_recv(RepStreamHandler(self.rep_stream, self.stop,
                                                 self.ping_handler))

    def run(self):
        """Sets up everything and starts the event loop."""
        self.setup()
        self.loop.start()

    def stop(self):
        """Stops the event loop."""
        self.loop.stop()

If you are going to start this process as a sub-process via start, make sure everything you instantiate in __init__ is pickle-able or it won’t work on Windows (Linux and Mac OS X use fork to create a sub-process and fork just makes a copy of the main process and gives it a new process ID. On Windows, there is no fork and the context of your main process is pickled and sent to the sub-process).

In setup, call super().setup() before you create a stream or you won’t have a loop instance for them. We call setup from run, because the context must be created within the new system process, which wouldn’t be the case if we called setup from __init__.

The stop method is not really necessary in this example, but it can be used to send stop messages to sub-processes when the main process terminates and to do other kinds of clean-up. You can also execute it if you except a KeyboardInterrupt after calling run.

MessageHandler — The Base Class for Message Handlers

A PyZMQ message handler can be any callable that accepts one argument—the list of message parts as byte objects. Hence, our MessageHandler class needs to implement __call__:

# exmaple_app/base.py

from zmq.utils import jsonapi as json


class MessageHandler(object):
    """
    Base class for message handlers for a :class:`ZMQProcess`.

    Inheriting classes only need to implement a handler function for each
    message type.

    """
    def __init__(self, json_load=-1):
        self._json_load = json_load

    def __call__(self, msg):
        """
        Gets called when a messages is received by the stream this handlers is
        registered at. *msg* is a list as return by
        :meth:`zmq.core.socket.Socket.recv_multipart`.

        """
        # Try to JSON-decode the index "self._json_load" of the message
        i = self._json_load
        msg_type, data = json.loads(msg[i])
        msg[i] = data

        # Get the actual message handler and call it
        if msg_type.startswith('_'):
            raise AttributeError('%s starts with an "_"' % msg_type)

        getattr(self, msg_type)(*msg)

As you can see, it’s quite simle. It just tries to JSON-load the index defined by self._json_load. We earlier defined, that the first element of the JSON-encoded message defines the message type (e.g., ping). If an attribute of the same name exists in the inheriting class, it is called with the remainer of the message.

You can also add logging or additional security measures here, but that is not necessary here.

RepStreamHandler — The Concrete Message Handler

This class inherits the MessageHandler I just showed you and is used in PongProc.setup. It defines a handler method for ping messages and the plzdiekthxbye stop message. In its __init__ it receives references to the rep_stream, PongProcs stop method and to the ping_handler, our actual application logic:

# example_app/pongproc.py

class RepStreamHandler(base.MessageHandler):
    """Handels messages arrvinge at the PongProc’s REP stream."""
    def __init__(self, rep_stream, stop, ping_handler):
        super().__init__()
        self._rep_stream = rep_stream
        self._stop = stop
        self._ping_handler = ping_handler

    def ping(self, data):
        """Send back a pong."""
        rep = self._ping_handler.make_pong(data)
        self._rep_stream.send_json(rep)

    def plzdiekthxbye(self, data):
        """Just calls :meth:`PongProc.stop`."""
        self._stop()

PingHandler – The Application Logic

The PingHandler contains the actual application logic (which is not much, in this example). The make_pong method just gets the number of pings sent with the ping message and creates a new pong message. The serialization is done by PongProc, so our Handler does not depend on PyZMQ:

# example_app/pongproc.py

class PingHandler(object):

    def make_pong(self, num_pings):
        """Creates and returns a pong message."""
        print('Pong got request number %s' % num_pings)

        return ['pong', num_pings]

Summary

Okay, that’s it for now. I showed you three ways to use PyZMQ. If you have a very simple process with only one socket, you can easily use its blocking recv methods. If you need more than one socket, I recommend using the event loop. And polling … you don’t want to use that.

If you decide to use PyZMQ’s event loop, you should separate the application logic from all the PyZMQ stuff (like creating streams, sending/receiving messages and dispatching them). If your application consists of more then one process (which is usually the case), you should also create a base class with shared functionality for them.

In the next part, I’m going to talk about how you can test your application.

Book review: NumPy 1.5 Beginner’s Guide

2011-11-21T18:55:00+01:00

I recently got the chance to review the book NumPy 1.5 Beginner’s Guide by Ivan Idris and published by Packt Publishing.

It covers many aspects of NumPy and also introduces SciPy as well as Matplotlib. The author includes a lot of examples and exercises and also shows the effects of some not-so-easy-to-understand functions using matplotlib graphs.

The book is easy to read, so you should make fast progress in learning NumPy. Overall, it’s a good read for NumPy beginners. Advanced NumPy users, who just want to look up how specific things work, are better of with NumPy’s documentation, though.

SimPy 2.2

2011-09-27T21:42:25+02:00

SimPy is a process-based discrete-event simulation library written in pure Python.

Ontje Lünsdorf and I have already contributed to prior version of SimPy and now have become members of the SimPy team.

In SimPy 2.2, we have restructured the package layout to be conform to the Hitchhiker’s Guide to packaging. We ported the the unit tests (524 test cases) to pytest. We improved and cleaned up parts of the documentation and finally, we fixed the behavior of Store._put—thanks to Johannes Koomer for the heads-up and the fix.

You can download SimPy from PyPI.

Building NumPy, SciPy & Matplotlib for Python 2.7 on Snow Leopard

2010-11-17T08:30:39+01:00

A few days ago I wrote about how to build SciPy for Python 2.7 on Mac OS 10.6 Snow Leopard.

Usually you want to install NumPy, SciPy and Matplotlib. After reading Installing SciPy/Mac OS X, the Matplotlib installation instructions and the HJBlog you might come to the conclusion, that it’s not trivial to build them on your own and that you better use the 32bit binaries for Python 2.6 or get them via MacPorts.

But actually it’s really easy. The only dependencies that you need to install are Xcode (for gcc and X11) and gfortran.

To simplify the the installation, I wrote a small Makefile that downloads all packages (except for Xcode) and builds/installs them:

Rename* the downloaded makefile to Makefile, open Terminal and cd to the diretory with the makefile.
Download gfortran and start the graphical installer:
```
$ make fortran
```
Download and install NumPy, SciPy and Matplotlib:
```
$ make
```
Run NumPy’s and SciPy’s test suite and delete the temporary build diretory:
```
$ make test clean
```

That’s how the makefile looks like:

# Download all packages to this directory.
TMP_DIR=./PYTMP

# Select Python version.
PYVERSION=2.7
PYTHON=python${PYVERSION}

# Package to download frm http://r.research.att.com/
FORTRANPACKAGE=gfortran-42-5664.pkg

# Select which versions of the packages you want to install.
NUMPYVERSION=1.5.0
SCIPYVERSION=0.8.0
MATPLOTLIBMAJORVERSION=1.0
MATPLOTLIBVERSION=${MATPLOTLIBMAJORVERSION}.0

# Normally, you shouldn’t need change this.
OSX_SDK_VER=10.6
ARCH_FLAGS=-arch i386 -arch x86_64

# Values for some environment variables. You shouldn’t need to change this.
MACOSX_DEPLOYMENT_TARGET=${OSX_SDK_VER}
CFLAGS="${ARCH_FLAGS} -I/usr/X11/include -I/usr/X11/include/freetype2 -isysroot /Developer/SDKs/MacOSX${OSX_SDK_VER}.sdk"
LDFLAGS="-Wall -undefined dynamic_lookup -bundle ${ARCH_FLAGS} -L/usr/X11/lib -syslibroot,/Developer/SDKs/MacOSX${OSX_SDK_VER}.sdk"
FFLAGS="${ARCH_FLAGS}"


default: all

# Create a temporary directory for the build process
_tmp_dir:
    mkdir -p ${TMP_DIR}

# Clean the the temporary build directory
clean:
    rm -rf ${TMP_DIR}

# Download gfortran and start the installer
fortran: _tmp_dir
    cd ${TMP_DIR} && \
    ${PYTHON} -c "import urllib; urllib.urlretrieve('http://r.research.att.com/${FORTRANPACKAGE}', '${FORTRANPACKAGE}')" && \
    open ${FORTRANPACKAGE}

# Download all required packages
fetch: _tmp_dir
    cd ${TMP_DIR} && \
    ${PYTHON} -c "import urllib; urllib.urlretrieve('http://sourceforge.net/projects/numpy/files/NumPy/${NUMPYVERSION}/numpy-${NUMPYVERSION}.tar.gz/download', 'numpy-${NUMPYVERSION}.tar.gz')" && \
    ${PYTHON} -c "import urllib; urllib.urlretrieve('http://sourceforge.net/projects/scipy/files/scipy/${SCIPYVERSION}/scipy-${SCIPYVERSION}.tar.gz/download', 'scipy-${SCIPYVERSION}.tar.gz')" && \
    ${PYTHON} -c "import urllib; urllib.urlretrieve('http://sourceforge.net/projects/matplotlib/files/matplotlib/matplotlib-${MATPLOTLIBMAJORVERSION}/matplotlib-${MATPLOTLIBVERSION}.tar.gz/download', 'matplotlib-${MATPLOTLIBVERSION}.tar.gz')"

# Extract, build and install NumPy
numpy:
    cd ${TMP_DIR} && \
    rm -rf numpy-${NUMPYVERSION} && \
    tar -xf numpy-${NUMPYVERSION}.tar.gz && \
    cd numpy-${NUMPYVERSION} && \
    export MACOSX_DEPLOYMENT_TARGET=${MACOSX_DEPLOYMENT_TARGET} && \
    export CFLAGS=${CFLAGS} && \
    export LDFLAGS=${LDFLAGS} && \
    export FFLAGS=${FFLAGS} && \
    ${PYTHON} setup.py build && \
    ${PYTHON} setup.py install && \
    cd .. && \
    ${PYTHON} -c "import numpy; print 'Installed NumPy', numpy.__version__"

# Extract, build and install SciPy
scipy:
    cd ${TMP_DIR} && \
    rm -rf scipy-${SCIPYVERSION} && \
    tar -xf scipy-${SCIPYVERSION}.tar.gz && \
    cd scipy-${SCIPYVERSION} && \
    export MACOSX_DEPLOYMENT_TARGET=${MACOSX_DEPLOYMENT_TARGET} && \
    export CFLAGS=${CFLAGS} && \
    export LDFLAGS=${LDFLAGS} && \
    export FFLAGS=${FFLAGS} && \
    ${PYTHON} setup.py build && \
    ${PYTHON} setup.py install && \
    cd .. && \
    ${PYTHON} -c "import scipy; print 'Installed SciPy', scipy.__version__"

# Extract, build and install Matplotlib
matplotlib:
    cd ${TMP_DIR} && \
    rm -rf matplotlib-${MATPLOTLIBVERSION} && \
    tar -xf matplotlib-${MATPLOTLIBVERSION}.tar.gz && \
    cd matplotlib-${MATPLOTLIBVERSION} && \
    export MACOSX_DEPLOYMENT_TARGET=${MACOSX_DEPLOYMENT_TARGET} && \
    export CFLAGS=${CFLAGS} && \
    export LDFLAGS=${LDFLAGS} && \
    ${PYTHON} setup.py build && \
    ${PYTHON} setup.py install && \
    cd .. && \
    ${PYTHON} -c "import matplotlib; print 'Installed Matplotlib', matplotlib.__version__"

# Fetch and build NumPy, SciPy and Matplotlib
all: fetch numpy scipy matplotlib
    echo "all done"

# Execute tests for NumPy and SciPy
test: _tmp_dir
    export MACOSX_DEPLOYMENT_TARGET=${MACOSX_DEPLOYMENT_TARGET} && \
    export CFLAGS=${CFLAGS} && \
    export LDFLAGS=${LDFLAGS} && \
    export FFLAGS=${FFLAGS} && \
    cd ${TMP_DIR} && \
    ${PYTHON} -c "import numpy; numpy.test('1', '10')" && \
    ${PYTHON} -c "import scipy; scipy.test('1', '10')"

I have only tested it on my local machine yet. Hope, it helps and you don’t run into any troubles.

* If you don’t rename the makefile, you must pass it’s name to make: make -f Makefile_nsm <target>.

Building SciPy on Snow Leopard with Python 2.7

2010-11-11T09:30:08+01:00

I recently had some struggle to build and install SciPy 0.8.0 on Mac OS X 10.6 Snow Leopard, but actually it’s quite easy.

I used the instructions on scipy.org and the HJBlog as source. Since there are builds of NumPy 1.5.0 for Python 2.7 available, you don’t need to install fftw and umfpack manually.

You only need to install gfortran from this site. Pick the latest build for Snow Leopard (e.g. this one).

Next, install NumPy with the disk image from SourceForge or with pip:

$ pip install numpy

To build and install SciPy, download the latest version from SourceForge and do the following:

$ tar -xf scipy-0.8.0.tar.gz
$ cd scipy-0.8.0
$ export CFLAGS="-arch i386 -arch x86_64"
$ export LDFLAGS="-Wall -undefined dynamic_lookup -bundle -arch i386 -arch x86_64"
$ export FFLAGS="-arch i386 -arch x86_64"
$ export MACOSX_DEPLOYMENT_TARGET=10.6
$ python setup.py build
$ python setup.py install

Everything in one command:

$ CFLAGS="-arch i386 -arch x86_64" LDFLAGS="-Wall -undefined dynamic_lookup -bundle -arch i386 -arch x86_64" FFLAGS="-arch i386 -arch x86_64" MACOSX_DEPLOYMENT_TARGET=10.6 python setup.py build install

I hope this works for you as well as it did for me.

Collectors 1.0

2010-06-24T19:52:59+02:00

It took me nearly three months to fix five small issues with the documentation. But now I finally released Collectors v1.0. :-)

You can read everything important in the RC1 posting.

Collectors 1.0-rc1

2010-04-03T10:39:15+02:00

Collectors made a huge jump von v0.1 to v1.0 over the last weeks, since we added lots of changes and consider what we’ve done as stable. If we don’t find any bugs, we’ll release the final v1.0 at the end of next week. The changes are:

[NEW] Documentation!
[NEW] Tests!
[NEW] Collectors can use different storage backends now
[NEW] Storage backend for PyTables
[NEW] Storage backend for MS Excel
[NEW] Collector.collect() as alias to Collector.__call__()
[CHANGE] Monitor is now called Collector
[CHANGE] Shortcut functions moved to shortcuts package

You can download and install Collectors via PyPI. If you find any bugs or have ideas for further enhancement, please let us know.

The Documentation can be found here.

Stefan Scherfke - python

Publishing to PyPI with a Trusted Publisher from GitLab CI/CD

Why should I want to use this?

How do Trusted Publishers work?

Environments

Security Considerations

Putting it all together

PyPI Settings

GitLab CI/CD Settings

Changes in Project Files

Typed Settings

What makes it different?

An Example

Project Status

Raise … from … in Python

Enter raise … from …

When the underlying cause is not important

Summary

Attrs, Dataclasses and Pydantic

Simple class definition

Class and attribute customistaion

Generated methods

Default values

Freezing and functional programming

Instantiation, validation and conversion

Metadata and schema generation

Documentation

Performance

Summary

Epilogue

Packaging Python inside your organization with GitLab and Conda

What to use?

Recipe organization

Linking against system packages

Recipe management

The update process

Example recipes

The macOS Dark Mode, your Terminal and Vim

How to mask an image with a smooth circle in PyQt5

Assertions and Exceptions

When do I use assert?

Validation of input parameters

What kind of exceptions should I raise?

TLDR

Getting started with devpi

Installation

Setting up the server

User management

Package indexes

Install packages

Upload packages

Congratulation!

Advanced asyncio testing

Requirements for our tests

Our first approach

Approach number two

Testing (asyncio) coroutines with pytest

aiomas – A library for multi-agent systems and RPC based on asyncio

What? But why?

How do I use it?

What’s the list of features?

What’s comming next?

Installation

Master(ing) Passwords

Available implementations

mpw 0.3

SimPy: Real-time simulations

SimPy: Shared Resources

The basic concept of resources

Resources and interrupts

Resources

Resource

PriorityResource

PreemptiveResource

Containers

Stores

SimPy: Process Interaction

Sleep until woken up

Waiting for another process to terminate

Interrupting another process

Enter `raise … from …`

When do I use `assert`?

Example usages for `Event`

Let time pass by: the `Timeout`