das-cli 1.5.272

DAS api client.

DAS CLI and Python SDK

das-cli is both:

• a command-line tool (das) for day-to-day DAS operations, and
• a Python package for scripts, automations, and integrations.

This README is organized around how people actually use the project:

1. CLI workflows — Authentication · Search and Read · Entries · Digital Objects · Downloads · Other CLI groups
2. Programmatic workflows — Pattern A · Pattern B
3. Manager-layer API reference — EntryManager · SearchManager · DownloadManager · DigitalObjectsManager

Install

Requirements

• Python 3.13+

Install from PyPI

pip install das-cli

Install from source

git clone https://git.nioz.nl/ict-projects/das-cli.git
cd das-cli
pip install -e .

Quickstart (CLI)

# 1) Authenticate (choose one)
das login --api-url https://your-das-instance/api
das login --api-url https://your-das-instance/api --api-key <your-api-key>

# 2) Search
das search entries --attribute Cores --query "name(*64*)" --max-results 10 --page 1

# 3) Get one entry
das entry get --code zb.b.1ub

CLI Usage

• Authentication
  • Username/password
  • API key
  • OAuth / Azure Entra
• Search and Read
• Entries
  • Create entries
  • Update entries
  • Logs and ownership
• Digital Objects
• Downloads
• Other CLI groups

Authentication

Username/password

das login --api-url https://your-das-instance/api --username your_user --password your_password

If --username or --password is omitted, the CLI prompts securely.

API key

das login --api-url https://your-das-instance/api --api-key <your-api-key>

Generate the key in DAS Web UI: My Settings -> API Keys.

OAuth / Azure Entra

das oauth configure --client-id <client-id> --authority https://login.microsoftonline.com/<tenant>/v2.0
das login --api-url https://your-das-instance/api --oauth

Search and Read

# Search with paging/sorting/formatting
das search entries --attribute Cores --query "name(*64*)" --max-results 10 --page 1 --sort-by Name --sort-order asc --format table

# Get details of one entry by ID
das search entry 6b0e68e6-00cd-43a7-9c51-d56c9c091123 --format json

# Query syntax help
das search help

Entries

# Get by code
das entry get --code ENT001

# Delete by code (with prompt)
das entry delete --code ENT001

# Delete by ID (no prompt)
das entry delete --id e1987f3e-7db7-11f0-a7e8-0edcca226f3d --force

Create entries

# From file (.json/.csv/.xlsx)
das entry create --attribute Cores c:\data\entries.json

# Single object from --data
das entry create --attribute Cores --data "{ 'Alias': 'test.entry.001', 'Event': '64PE428-02PC' }"

# Bulk from --data
das entry create --attribute Cores --data "[{ 'Alias': 'a1' }, { 'Alias': 'a2' }]"

Update entries

# Single by code
das entry update --code ENT001 --data "{ 'Comment': 'Updated by CLI' }"

# Single by id
das entry update --id e1987f3e-7db7-11f0-a7e8-0edcca226f3d --data "{ 'Comment': 'Updated by id' }"

# Bulk from file (each item should include Id/Code)
das entry update c:\data\entries-to-update.json

Logs and ownership

# Audit logs
das entry logs --code ENT001 --max-results 50 --skip 0 --format table

# Change owner
das entry chown --user alice --code ENT001 --code ENT002

Digital Objects

# Upload + link to entry
das entry upload-digital-object --entry-code ENT001 --type Dataset --description "CTD raw" c:\data\ctd.zip

# Link existing digital objects
das entry link-digital-objects --entry-code ENT001 -d DO001 -d DO002

# Unlink
das entry link-digital-objects --entry-code ENT001 -d DO002 --unlink

# Direct download by codes
das entry direct-download-digital-object --entry-code ENT001 --digital-object-code DO001 --out C:\Downloads\

Downloads

# Create request for all files from entry
das download request --entry ENT001 --name "My request"

# Create request selecting files
das download request --entry ENT001 --file FILE001 --file FILE002

# List your requests
das download my-requests --format table

# Download completed bundle
das download files 6b0e68e6-00cd-43a7-9c51-d56c9c091123 --out C:\Downloads

# Delete request
das download delete-request 6b0e68e6-00cd-43a7-9c51-d56c9c091123

Other CLI groups

# Attribute lookup
das attribute get --name "Cores"
das attribute get-id "Cores"
das attribute get-name 126

# Cache
das cache list
das cache clear "attributes"
das cache clear-all

# Config
das config ssl-verify true
das config ssl-status
das config show
das config reset --force

Programmatic Usage (Python)

Two common patterns are supported.

• Pattern A: login once with CLI, then use managers in Python
• Pattern B: authenticate in Python with Das
• Managers — full API reference below:
  • EntryManager
  • SearchManager
  • DownloadManager
  • DigitalObjectsManager

Pattern A: login once with CLI, then use managers in Python

from das.managers.entries_manager import EntryManager
from das.managers.search_manager import SearchManager

entries = EntryManager()
search = SearchManager()

core = entries.get(code="zb.b.1ub")
results = search.search_entries(attribute="Cores", query="name(*64*)", max_results=5)

Pattern B: authenticate in Python with Das

from das.app import Das
from das.managers.entries_manager import EntryManager

# Username/password
client = Das("https://your-das-instance/api")
client.authenticate("your_user", "your_password")

# Or API key
# client = Das("https://your-das-instance/api")
# client.authenticate_with_api_key("your-api-key")

entry_manager = EntryManager()
entry = entry_manager.get(code="zb.b.1ub")
print(entry.get("Name"))

Manager Layer API Reference (Domain Layer)

Managers are the recommended Python API. They translate user-friendly field names, codes, and query syntax into the lower-level DAS API calls.

Prerequisite: authenticate first (see Programmatic Usage), then instantiate any manager:

from das.managers.entries_manager import EntryManager

entries = EntryManager()  # reads API URL + token from stored config

All managers raise ValueError when the API URL is not configured or when required parameters are missing.

• EntryManager
  • get / get_entry
  • create
  • update
  • delete
  • create_from_file
  • chown
  • get_entry_logs
  • get_direct_relations
• SearchManager
  • search_entries
• DownloadManager
  • create_download_request
  • get_my_requests
  • delete_download_request
  • download_files
  • save_download
• DigitalObjectsManager
  • upload_digital_object
  • link_existing_digital_objects
  • download_digital_object
  • direct_download_digital_object
• Complete end-to-end script

---

EntryManager

Import: from das.managers.entries_manager import EntryManager

Method	Returns
get(code=..., id=...)	dict — entry with display field names
get_entry(entry_id)	dict — same as get(id=...)
create(attribute, entry=..., entries=...)	list[dict] — per-item status with id, code, status
update(id=..., code=..., entry=..., entries=...)	list[dict] — per-item status
delete(id=..., code=...)	bool
create_from_file(attribute, file_path)	None
chown(user_name, entry_code_list)	API response
get_entry_logs(entry_id=..., entry_code=..., ...)	dict with totalCount and items
get_direct_relations(entry_code=..., entry_id=..., ...)	dict with relation items

get / get_entry — fetch one entry

entries = EntryManager()

# By code (returns display field names like "Name", "Code", "Event", etc.)
entry = entries.get(code="zb.b.1ub")
print(entry["Name"], entry["Code"])

# By ID (GUID)
entry = entries.get(id="6b0e68e6-00cd-43a7-9c51-d56c9c091123")

# Convenience alias — same result as get(id=...)
entry = entries.get_entry("6b0e68e6-00cd-43a7-9c51-d56c9c091123")

create — create one or many entries

entries = EntryManager()

# Single entry (field keys can be display names or column names)
result = entries.create(attribute="Cores", entry={
    "Alias": "test.entry.001",
    "Event": "64PE428-02PC",
    "Number": 1,
    "Number Alias": "001",
    "Start Depth": 576,
    "End Depth": 600,
    "Comment": "Created from Python",
})
# result -> [{"id": "<guid>", "code": "zb.b.xxx", "status": "success"}]
new_id = result[0]["id"]

# Bulk create
results = entries.create(attribute="Cores", entries=[
    {"Alias": "bulk.001", "Event": "64PE428-02PC", "Number": 1},
    {"Alias": "bulk.002", "Event": "64PE428-02PC", "Number": 2},
])
for item in results:
    print(item.get("id"), item.get("status"), item.get("error"))

# Entry with relation fields (comma-separated codes or GUIDs)
entries.create(attribute="Cruise", entry={
    "Cruise Code": "TEST-Cruise-001",
    "Project(s)": "3d.b.q,3d.b.r",   # multiple project codes
    "Vessel": "Algoa",
    "Start Date": "2026-01-01",
    "End Date": "2026-01-02",
})

update — update one or many entries

entries = EntryManager()

# Partial update by code (only changed fields needed)
entries.update(code="zb.b.0", entry={
    "Availability": "pc.b.c",
    "Number": 2,
    "Comment": "Updated from Python",
})

# Partial update by ID
entries.update(id="c4c6dff6-5ccb-11f0-a78d-a84a63c8db73", entry={
    "description": "New description text",
    "jobtitle": "Head of National Marine Facilities",
})

# Bulk update (each item must include Code or Id)
entries.update(entries=[
    {"Code": "ENT001", "Comment": "Updated 1"},
    {"Code": "ENT002", "Comment": "Updated 2"},
])

delete — delete one entry

entries = EntryManager()

entries.delete(id="6b0e68e6-00cd-43a7-9c51-d56c9c091123")  # -> True
entries.delete(code="zb.b.1ub")                             # -> True

create_from_file — bulk create from file

entries = EntryManager()

entries.create_from_file(attribute="Cores", file_path=r"c:\data\entries.json")
entries.create_from_file(attribute="Cores", file_path=r"c:\data\entries.csv")
entries.create_from_file(attribute="Cores", file_path=r"c:\data\entries.xlsx")

Supported formats: .json (list of objects), .csv (one row per entry), .xlsx (one row per entry).

chown — transfer ownership

entries = EntryManager()

result = entries.chown(
    user_name="alice",
    entry_code_list=["ENT001", "ENT002"],
)

get_entry_logs — audit / change history

entries = EntryManager()

# By code
logs = entries.get_entry_logs(entry_code="ENT001", max_result_count=50, skip_count=0)

# By ID
logs = entries.get_entry_logs(entry_id="d7b99f36-5ccb-11f0-87f3-a84a63c8db73")

print(logs["totalCount"])
for log in logs["items"]:
    print(log["fieldName"], log["valueFrom"], "->", log["valueTo"], log["modifierUserName"])

get_direct_relations — related entries

entries = EntryManager()

relations = entries.get_direct_relations(entry_code="6.b.sy")
for rel in relations["items"]:
    print(rel["id"], rel.get("displayname"))

---

SearchManager

Import: from das.managers.search_manager import SearchManager

Method	Returns
search_entries(attribute, query, max_results, page, sort_by, sort_order, includeRelationsId)	dict with items and totalCount

Query and sort field names are automatically converted from display names to internal column names.

search_entries — search within an attribute

search = SearchManager()

# Basic search
results = search.search_entries(
    attribute="Cores",
    query="name(*64*)",
    max_results=10,
    page=1,
)
print(results["totalCount"])
for item in results["items"]:
    print(item.get("Name"), item.get("Code"))

# Composite query (AND / OR)
results = search.search_entries(
    attribute="Cores",
    query="name(64PE320-03BC#01) or code(zb.b.1ub)",
    max_results=5,
    sort_by="Name",
    sort_order="asc",
)

# Search all entries in an attribute (empty query)
results = search.search_entries(
    attribute="Cores",
    query="",
    max_results=1000,
    page=1,
)

# Find stations by cruise code
stations = search.search_entries(
    attribute="Station",
    query="Cruise(64PE514)",
    max_results=5,
    sort_by="Code",
)

# Wildcard search on display name
projects = search.search_entries(
    attribute="Cruise",
    query="displayname(*64PE514*)",
    max_results=5,
    sort_by="displayname",
)

# Paginate to page 2
page2 = search.search_entries(
    attribute="Cores",
    query="name(*64*)",
    max_results=50,
    page=2,
)

---

DownloadManager

Import: from das.managers.download_manager import DownloadManager

Method	Returns
create_download_request(request_data)	str (request ID) or dict with errors
get_my_requests()	dict or list of download requests
delete_download_request(request_id)	API response
download_files(request_id)	requests.Response (streaming)
save_download(request_id, output_path, overwrite)	str — saved file path

The request_data dict uses entry codes as keys. Values are lists of digital-object codes; an empty list means all files from that entry.

create_download_request — request files for download

downloads = DownloadManager()

# Specific files from one entry
request_id = downloads.create_download_request({
    "name": "My download",
    "zb.b.dc": ["h.b.z0pg", "h.b.s3pg"],
})

# All files from one entry (empty list)
request_id = downloads.create_download_request({
    "name": "All files",
    "zb.b.lu": [],
})

# Mixed: selected files from one entry, all files from another
request_id = downloads.create_download_request({
    "name": "Mixed download",
    "zb.b.dc": ["h.b.z0pg", "h.b.s3pg"],
    "zb.b.lu": [],
})

# Handle validation errors
result = downloads.create_download_request({"name": "Bad", "INVALID": []})
if isinstance(result, dict) and "errors" in result:
    for err in result["errors"]:
        print(err)

get_my_requests — list your download requests

downloads = DownloadManager()

requests = downloads.get_my_requests()
items = requests.get("items", []) if isinstance(requests, dict) else requests
for req in items:
    print(req["id"], req.get("status"), len(req.get("files", [])))

delete_download_request — remove a request

downloads = DownloadManager()

downloads.delete_download_request("6b0e68e6-00cd-43a7-9c51-d56c9c091123")

download_files — get streaming response

downloads = DownloadManager()

response = downloads.download_files("6b0e68e6-00cd-43a7-9c51-d56c9c091123")

# Manual save from stream
with open(r"C:\Downloads\bundle.zip", "wb") as f:
    for chunk in response.iter_content(chunk_size=8192):
        if chunk:
            f.write(chunk)

save_download — download and save in one step

downloads = DownloadManager()

# Save to directory (filename taken from server response)
path = downloads.save_download(
    request_id="6b0e68e6-00cd-43a7-9c51-d56c9c091123",
    output_path=r"C:\Downloads",
)

# Save to explicit file path, overwrite if exists
path = downloads.save_download(
    request_id="6b0e68e6-00cd-43a7-9c51-d56c9c091123",
    output_path=r"C:\Downloads\bundle.zip",
    overwrite=True,
)

---

DigitalObjectsManager

Import: from das.managers.digital_objects_manager import DigitalObjectsManager

Method	Returns
upload_digital_object(entry_code, file_description, digital_object_type, file_path)	str — digital object ID
link_existing_digital_objects(entry_code, digital_object_code_list, is_unlink)	bool
download_digital_object(output_path, digital_object_id, digital_object_code)	str — saved file path
direct_download_digital_object(...)	str — saved file path

upload_digital_object — upload file and link to entry

digital_objects = DigitalObjectsManager()

do_id = digital_objects.upload_digital_object(
    entry_code="zb.b.f7",
    file_description="CTD raw data",
    digital_object_type="Dataset",
    file_path=r"c:\data\ctd.zip",
)
print(do_id)  # GUID of the new digital object

link_existing_digital_objects — link or unlink by codes

digital_objects = DigitalObjectsManager()

# Link
digital_objects.link_existing_digital_objects(
    entry_code="4d.b.1l",
    digital_object_code_list=["h.b.ac0j"],
    is_unlink=False,
)

# Unlink
digital_objects.link_existing_digital_objects(
    entry_code="4d.b.1l",
    digital_object_code_list=["h.b.ac0j"],
    is_unlink=True,
)

download_digital_object — download by digital object code or ID

digital_objects = DigitalObjectsManager()

# By code (output_path can be a directory or a file path)
path = digital_objects.download_digital_object(
    output_path=r"c:\temp\",
    digital_object_code="h.b.8j2h",
)

# By ID
path = digital_objects.download_digital_object(
    output_path=r"c:\temp\file.bin",
    digital_object_id="b47c3de3-d754-4a25-8528-c4fd88e1fdbb",
)

direct_download_digital_object — download from a specific entry

digital_objects = DigitalObjectsManager()

# By entry code + digital object code
path = digital_objects.direct_download_digital_object(
    entry_code="4d.b.1l",
    digital_object_code="h.b.ac0j",
    output_path=r"c:\temp\",
)

# By IDs (when you already have them)
path = digital_objects.direct_download_digital_object(
    entry_id="e1987f3e-7db7-11f0-a7e8-0edcca226f3d",
    attribute_id=126,
    digital_object_id="cf335b17-80e3-4151-aa80-40b69fe0567e",
    output_path=r"c:\temp\file.bin",
)

---

Complete end-to-end script

from das.app import Das
from das.managers.entries_manager import EntryManager
from das.managers.search_manager import SearchManager
from das.managers.download_manager import DownloadManager
from das.managers.digital_objects_manager import DigitalObjectsManager

# 1. Authenticate
client = Das("https://your-das-instance/api")
client.authenticate("your_user", "your_password")

# 2. Instantiate managers
entries = EntryManager()
search = SearchManager()
downloads = DownloadManager()
digital_objects = DigitalObjectsManager()

# 3. Search
hits = search.search_entries(attribute="Cores", query="name(*64*)", max_results=5)
print(f"Found {hits['totalCount']} cores")

# 4. Read one entry
core = entries.get(code="zb.b.1ub")
print(core["Name"])

# 5. Create a new entry
created = entries.create(attribute="Cores", entry={
    "Alias": "api.test.001",
    "Event": "64PE428-02PC",
    "Number": 1,
})
print(f"Created {created[0]['code']}")

# 6. Upload a file
do_id = digital_objects.upload_digital_object(
    entry_code=created[0]["code"],
    file_description="Test file",
    digital_object_type="Dataset",
    file_path=r"c:\data\sample.zip",
)

# 7. Request download and save
request_id = downloads.create_download_request({
    "name": "API test download",
    created[0]["code"]: [],
})
saved = downloads.save_download(request_id, output_path=r"C:\Downloads")
print(f"Saved to {saved}")

# 8. Clean up
entries.delete(id=created[0]["id"])
downloads.delete_download_request(request_id)

Configuration and Environment Variables

You can configure values through CLI commands and/or .env.

# Connection
DAS_API_URL=https://api.das-dev.nioz.nl
DAS_API_KEY=
VERIFY_SSL=True

# Test credentials (for integration tests)
API_URL=https://api.das-dev.nioz.nl
USER_NAME=
USER_PASSWORD=

# AI feature
OPENAI_API_KEY=
OPEN_AI_MODEL_ID=gpt-4o-mini

• DAS_API_URL: default API URL for app/managers when no explicit URL is provided.
• DAS_API_KEY: API key auth (takes precedence over stored token in auth layer).
• VERIFY_SSL: TLS verification (True/False).
• API_URL, USER_NAME, USER_PASSWORD: used by integration tests.

Unit Tests and Integration Tests

Run complete suite:

python run_tests.py

Run a single test module:

python -m pytest tests/entries_manager_test.py

The tests are a good source of realistic payload examples, especially:

• tests/entries_manager_test.py
• tests/search_manager_test.py
• tests/download_manager_test.py
• tests/digital_object_manager_test.py

Troubleshooting

• Authentication errors:
  • verify the URL includes your DAS API base (often ending in /api).
  • run das login again to refresh stored auth.
• SSL errors in local/dev environments:
  • temporarily disable with das config ssl-verify false (avoid in production).
• PowerShell quoting:
  • prefer double quotes around query arguments containing *, ;, and parentheses.
• No API URL configured:
  • set DAS_API_URL, pass URL to Das(...), or run das login --api-url ....

Contributing

Contributions are welcome through pull requests.

For bugs or feature requests, use the project issue tracker:

• DAS CLI issue tracker

License

MIT License