PySLL

SDKs and example code for accessing the Constellation APIs that power Emerald Cloud Lab.

The constellation API

Detailed documentation about the Constellation API can be found at www.emeraldcloudlab.com/internal-developers-api.

Environment Variables

If you are setting these in your .bash_profile (or similar), make sure to export them or they won't be available to os.environ.

Standard Credentials

• CONSTELLATION_AUTH_TOKEN (Optional)
• CONSTELLATION_USERNAME (Required if no token)
• CONSTELLATION_PASSWORD (Required if no token)

Test Credentials

To run the test suite, use the PYTEST prefix. The login has to be service+manifold@emeraldcloudlab.com since some tests expect this user ID.

• PYTEST_CONSTELLATION_AUTH_TOKEN (Optional)
• PYTEST_CONSTELLATION_USERNAME (Required if no token)
• PYTEST_CONSTELLATION_PASSWORD (Required if no token)

Example for tests:

export PYTEST_CONSTELLATION_USERNAME="your_username"
export PYTEST_CONSTELLATION_PASSWORD="your_password"

Development Setup

After cloning the repository, set up the development environment:

# Install dependencies
uv sync --dev

# Install git hooks for pre-commit and pre-push
uv run pre-commit install
uv run pre-commit install --hook-type pre-push

The pre-commit hooks will automatically:

• Run ruff formatting and linting on commit
• Run pyright type checking on commit
• Run the full test suite on push

Quick start

Install with pip:

pip install pysll

To use the SDK:

>>> from pysll import Constellation
>>> from pysll.models import Object
>>> client = Constellation()

To login:

>>> client.login("scientist@science.com", "myAwesomePassword")

Alternatively, load credentials automatically from environment variables using from_env():

>>> client = Constellation.from_env()

This reads CONSTELLATION_AUTH_TOKEN if set, otherwise falls back to CONSTELLATION_USERNAME and CONSTELLATION_PASSWORD. An optional prefix argument lets you namespace the variables (e.g., prefix="PYTEST" reads PYTEST_CONSTELLATION_USERNAME, etc.).

To get information about the current user once you are logged in:

>>> me = client.me()
>>> print(me)
{'Email': 'scientist@science.com', 'EmailAddress':'scientist@science.com', 'Id': 'id:abc123', 'Type': 'Object.User', 'Username': 'scientist'}

To download information from an object:

>>> client.download(Object(me["Id"]), ["Name", "Email"])
["scientist", "scientist@science.com"]

To search for objects of a specific type:

>>> client.search("Object.User.Emerald.Administrator", "")

Searching with conditions

You can perform more specific searches by providing an SLL query string:

>>> client.search(
...     "Object.SupportTicket.Operations",
...     'Status="Active" AND Priority="High"'
... )

You can also limit the number of results:

>>> client.search("Object.SupportTicket.Operations", "", max_results=5)

Downloading data

You may perform a simple single field download like:

>>> client.download(Object("id:BYDOjvG4l3Ol"), "ColumnOrientation")
'Forward'

You may download multiple fields in a single download like:

>>> client.download(Object("id:BYDOjvG4l3Ol"), ["SeparationMode", "ColumnOrientation"])
['ReversePhase', 'Forward']

You may download from multiple objects in a single download like:

>>> client.download([Object("id:o1k9jAkRM794"), Object("id:L8kPEjkw47jw")], "ColumnOrientation")
['Forward', 'Forward']

And finally, you may download multiple fields from multiple objects in a single download like:

>>> client.download([Object("id:o1k9jAkRM794"), Object("id:L8kPEjkw47jw")], ["SeparationMode", "ColumnOrientation"])
[['ReversePhase', 'Forward'], ['ReversePhase', 'Forward']]

You may also traverse links within downloads, like:

>>> client.download(Object("id:BYDOjvG4l3Ol"), "Instrument[Model[Name]]")
'Waters Acquity UPLC H-Class ELS with Pre-Column Heater'

You can also download all of the fields on an object by not specifying a field. For example:

>>> client.download(Object("id:Z1lqpMzvkGMV"))
{'type': 'Object.User.Emerald.Developer', 'id': 'id:Z1lqpMzvkGMV'....}

Or via the "All" implicit field:

>>> client.download(Object("id:Z1lqpMzvkGMV"), "All")
{'type': 'Object.User.Emerald.Developer', 'id': 'id:Z1lqpMzvkGMV'....}

Note that in this case, the results will be a dictionary mapping field name to field value

Dealing with types

There are a number of different ways to interpret field values based off the type of data stored in the object. String, integer, and real fields are mapped to their corresponding python types - for example:

>>> client.download(Object("id:BYDOjvG4l3Ol"), ["SeparationMode", "InjectionIndex"])
['ReversePhase', 28]

Link fields will return objects, which you can chain downloads off of (although note that traversals will be much faster):

>>> client.download(Object("id:BYDOjvG4l3Ol"), "Instrument")
Object[Instrument[HPLC, "id:wqW9BP4ARZVw"]
>>> client.download(client.download(Object("id:BYDOjvG4l3Ol"), "Instrument"), "Name")
'Galadriel'
>>> client.download(Object("id:BYDOjvG4l3Ol"), "Instrument[Name]")
'Galadriel'

Date fields will be converted to native python datetime objects:

>>> client.download(Object("id:BYDOjvG4l3Ol"), "DateCreated")
datetime.datetime(2022, 1, 9, 23, 44, 31, 746154)

Quantity arrays will be converted to python variable unit objects:

>>> client.download(Object("id:BYDOjvG4l3Ol"), "Scattering")
[[0.0 Minutes, -87.528984 IndependentUnit[Lsus]], [0.016667 Minutes, -96.701614 IndependentUnit[Lsus]], [0.033333 Minutes, -43.93272 IndependentUnit[Lsus]], [0.05 Minutes, -132.207855 IndependentUnit[Lsus]]...

which you may manipulate to get their values and units:

>>> scattering_info = client.download(Object("id:BYDOjvG4l3Ol"), "Scattering")
>>> len(scattering_info)
361
>>> scattering_info[0]
[0.0 Minutes, -87.528984 IndependentUnit[Lsus]]
>>> scattering_info[0][0]
0.0 Minutes
>>> scattering_info[0][0].value
0.0
>>> scattering_info[0][0].unit
'Minutes'

Blob refs will be downloaded and automatically parsed in the same way:

>>> client.download(Object("id:BYDOjvG4l3Ol"), "Absorbance")
[[0.0 'Minutes', 0.0 'Milli' 'AbsorbanceUnit'], [0.0008333333535119891 'Minutes', 0.0 'Milli' 'AbsorbanceUnit']...

Additionally, you can download multiple fields that have different units the same as you would download other fields. For example:

>>> client.download(Object("id:O81aEB16GlJ1"), "Composition")
[[4.977777777777776 Times[Power["Liters", -1], "Milligrams"], Object[Model[Molecule, "id:E8zoYvN6m61A"]], [75.11111111111111 IndependentUnit["VolumePercent"], Object[Model[Molecule, "id:vXl9j57PmP5D"]]]

Finally, you can download association fields and they will be automatically translated into python structures. For example:

>>> client.download(Object("id:XnlV5jKZwmp3"), "ResolvedOptions")['Instrument']
Object[Instrument[HPLC, "id:wqW9BP4ARZVw"]]

Download Files

Files are controlled via the auto_download_cloud_files flag to the download function. By default, they will be returned as objects and not downloaded.

For example:

>>> client.download(Object("id:BYDOjvG4l3Ol"), "DataFile")
Object[EmeraldCloudFile, "id:9RdZXv1jDAZ6"]

These may be manually downloaded via:

>>> client.download_cloud_file(client.download(Object("id:BYDOjvG4l3Ol"), "DataFile"))
'/var/folders/j_/ftdn14ms37s40j2z0h1wzxbw0000gn/T/tmp6krhb8lp/Absorbance Raw File.bin_absorbancefile'

or, it is possible to automatically download them by using the auto_download_cloud_files flag of download:

>>> data_file = client.download(Object("id:BYDOjvG4l3Ol"), "DataFile", auto_download_cloud_files=True)
>>> data_file.local_path
'/var/folders/j_/ftdn14ms37s40j2z0h1wzxbw0000gn/T/tmp6krhb8lp/Absorbance Raw File_1.bin_absorbancefile'

The format of these files can often change, but the sdk is pretty smart about interpreting them. Once you have downloaded the file, you can have the sdk attempt to parse it into python structs via the following:

>>> data_file = client.download(Object("id:BYDOjvG4l3Ol"), "DataFile", auto_download_cloud_files=True)
>>> from pysll.models import ConstellationFieldParser
>>> ConstellationFieldParser().parse_local_file(data_file.local_path)
[[0.0 'Minutes', 273.0 'Nanometers', 0.0 'Milli' 'AbsorbanceUnit'], [0.0008333333535119891 'Minutes', 273.0 'Nanometers', 0.0 'Milli' 'AbsorbanceUnit']...

If the field parser is unable to parse the file, it will return None.

Development

This project uses uv for dependency management and pre-commit for code quality checks.

Setup

1. Install uv (if not already installed):

   curl -LsSf https://astral.sh/uv/install.sh | sh
   

2. Install dependencies:

   uv sync --all-groups
   

   This installs all project dependencies including dev dependencies (pytest, ruff, pyright, etc.).

3. Set up pre-commit hooks:

   uv run pre-commit install
   

   This installs git hooks that automatically run linters and formatters before each commit.

Running Tests

Run the full test suite:

uv run pytest

Run tests in parallel (faster):

uv run pytest -n auto

Run tests with coverage:

uv run pytest --cov=pysll --cov-report=term-missing

Run specific tests:

uv run pytest tests/test_constellation.py
uv run pytest tests/test_constellation.py::test_login

Code Quality

Run pre-commit checks manually on all files:

uv run pre-commit run --all-files

Run specific checks:

# Format code with ruff
uv run ruff format

# Lint code with ruff
uv run ruff check

# Type checking with pyright
uv run pyright

Adding Dependencies

Use uv add to add new dependencies (never edit pyproject.toml directly):

# Add a runtime dependency
uv add package-name

# Add a dev dependency
uv add --group dev package-name

# Add an optional dependency
uv add --optional extra-name package-name

Uploading data

The upload function allows you to create new objects or update existing ones.

Create a new object

To create a new object, pass None as the object_id:

>>> client.upload(
...     "Object.SupportTicket.Operations",
...     None,
...     {
...         "Name": "New Support Request",
...         "Description": "Calculated results are missing",
...         "Status": "Open"
...     }
... )

Update an existing object

To update an existing object, provide its ID:

>>> client.upload(
...     "Object.SupportTicket.Operations",
...     "id:ticket-123",
...     {"Status": "InProgress"}
... )

Advanced Upload Examples

Linking to other objects:

>>> from pysll.utils import create_one_way_link
>>> client.upload(
...     "Object.SupportTicket.Operations",
...     "id:ticket-123",
...     {
...         "Assignee": create_one_way_link("Object.User.Emerald.Developer", "id:user-456")
...     }
... )

Appending to a list field:

>>> client.upload(
...     "Object.Sample",
...     "id:sample-123",
...     {"Append[Tags]": ["tag1", "tag2"]}
... )

Setting a quantity field:

>>> client.upload(
...     "Object.Sample",
...     None,
...     {"Volume": 'Quantity[0.5, "Liters"]'}
... )

Batch Uploading

Use upload_many to create or update multiple objects in a single request. Each element is a (object_type, object_id, fields) tuple — pass None for object_id to create a new object.

Create multiple objects at once:

>>> results = client.upload_many([
...     ("Object.Sample", None, {"Name": "Sample A"}),
...     ("Object.Sample", None, {"Name": "Sample B"}),
...     ("Object.Sample", None, {"Name": "Sample C"}),
... ])
>>> [r["resolved_object"]["id"] for r in results]
['id:abc', 'id:def', 'id:ghi']

Mix creates and updates in one call:

>>> results = client.upload_many([
...     ("Object.Sample", None, {"Name": "New Sample"}),
...     ("Object.Sample", "id:existing-123", {"Name": "Updated Sample"}),
... ])

Uploading Files

You can upload local files to the cloud:

>>> client.upload_cloud_file("path/to/my_data.txt")

Notebook Context

You can specify a notebook context for your uploads:

>>> with client.notebook("id:notebook-123"):
...     client.upload("Object.Example.Data", None, {"Name": "Contextual Data"})

Object History

Use object_log() to retrieve the change history for one or more objects:

>>> from datetime import datetime
>>> client.object_log(
...     object_ids=["id:abc123"],
...     start_date=datetime(2024, 1, 1),
...     max_results=100
... )

You can also filter by type instead of by specific IDs:

>>> client.object_log(types=["Object.SupportTicket.Operations"], max_results=50)

Type Introspection

Use get_type() to retrieve field definitions for any Constellation type:

>>> client.get_type("Object.SupportTicket.Operations")

This returns the field names, formats, and other metadata for the type, which is useful when exploring the schema or building dynamic queries.

Optional: NumPy Integration

If you have NumPy installed, pysll can use it for more efficient handling of large numerical arrays. Enable it by setting the PYSLL_USE_NUMPY environment variable:

export PYSLL_USE_NUMPY=true

Install with NumPy support:

pip install pysll[numpy]
# or via uv:
uv add pysll --optional numpy