REST API: examples

The scripts use the v3 interface of the REST API, which describes sign-in, permissions and best practices. The explanatory text around the scripts exists in German only: REST-API: Beispiele.

A minimal client

Ruby

# frozen_string_literal: true

require 'json'
require 'net/http'
require 'uri'

# Minimal client for the Librario v3 REST API, using only the standard library.
#
# Every v3 request carries an OAuth 2.1 bearer token. The token is bound to a single audience —
# exactly "<base_url>/api/v3" — so a token minted for one Librario host is rejected by any other.
# See oauth_pkce_login.rb for how to obtain one.
#
#   client = LibrarioClient.new(
#     base_url: 'https://acme.mylibrar.io',
#     access_token: ENV.fetch('LIBRARIO_ACCESS_TOKEN')
#   )
#   client.whoami
class LibrarioClient
  # Raised on any non-2xx response. The API answers with { "error": { "code", "message" } }.
  class Error < StandardError
    attr_reader :status, :code

    def initialize(status, body)
      @status = status
      @code = body.dig('error', 'code')
      super("HTTP #{status} #{@code}: #{body.dig('error', 'message') || body}")
    end
  end

  def initialize(base_url:, access_token:)
    @base_url = base_url.to_s.chomp('/')
    @access_token = access_token
  end

  # The authenticated user. The cheapest way to check that a token works.
  def whoami = get('/users/me')

  # Filters are the flat query parameters of GET /publications: doi, isbn, issn, arxiv_id, pmid,
  # author, category, year, query (full text), ... The typed `identifiers` array is the *response*
  # shape; you still search by the flat parameter.
  def publications(**filters) = get('/publications', **filters)

  # Yields every publication matching the filters, page by page. List endpoints cap out at
  # 25 records per page.
  def each_publication(per_page: 25, **filters, &)
    return to_enum(:each_publication, per_page:, **filters) unless block_given?

    page = 1
    loop do
      records = publications(page:, per_page:, **filters).fetch('records')
      records.each(&)
      break if records.size < per_page

      page += 1
    end
  end

  def create_publication(attributes) = post('/publications', attributes)
  def publication(id) = get("/publications/#{id}")
  def delete_publication(id) = delete("/publications/#{id}")

  def users(**filters) = get('/users', **filters)
  def update_user(id, attributes) = patch("/users/#{id}", attributes)

  # Presigns a direct-to-S3 upload. Returns { "url", "fields" }; `fields["key"]` becomes the
  # asset's file id once the "cache/" prefix is stripped.
  def presign_asset(filename:, type:) = get('/assets/presign', filename:, type:)

  # Presigns a direct-to-S3 cover-image upload. Returns { "url", "fields" };
  # `fields["key"]` becomes the publication's cover id once the "cache/" prefix is
  # stripped. Set the resulting { id:, storage: "cache", metadata: } hash as `cover`
  # on a publication create/update to attach it.
  def presign_cover(filename:, type:) = get('/publications/cover/presign', filename:, type:)

  def create_asset(attributes) = post('/assets', attributes)

  private

  def get(path, **query) = send_request(Net::HTTP::Get, path, query:)

  def send_request(verb, path, query: {}, body: nil)
    uri = URI.parse("#{@base_url}/api/v3#{path}")
    uri.query = URI.encode_www_form(query.compact) unless query.empty?

    request = verb.new(uri)
    request['Authorization'] = "Bearer #{@access_token}"
    request['Accept'] = 'application/json'
    if body
      request['Content-Type'] = 'application/json'
      request.body = JSON.generate(body)
    end

    response = Net::HTTP.start(uri.hostname, uri.port, use_ssl: uri.scheme == 'https') do |http|
      http.request(request)
    end
    parse(response)
  end

  def parse(response)
    parsed = response.body.to_s.empty? ? {} : JSON.parse(response.body)
    raise Error.new(response.code.to_i, parsed) unless response.is_a?(Net::HTTPSuccess)

    parsed
  end

  def post(path, body) = send_request(Net::HTTP::Post, path, body:)
  def delete(path) = send_request(Net::HTTP::Delete, path)
  def patch(path, body) = send_request(Net::HTTP::Patch, path, body:)
end

Python

"""Minimal client for the Librario v3 REST API.

Every v3 request carries an OAuth 2.1 bearer token. The token is bound to a single audience --
exactly ``<base_url>/api/v3`` -- so a token minted for one Librario host is rejected by any other.
See ``oauth_pkce_login.py`` for how to obtain one.

    client = LibrarioClient(base_url="https://acme.mylibrar.io", access_token=os.environ["LIBRARIO_ACCESS_TOKEN"])
    client.whoami()
"""

from __future__ import annotations

from collections.abc import Iterator
from typing import Any

import httpx


class LibrarioError(Exception):
    """Raised on any non-2xx response. The API answers with ``{"error": {"code", "message"}}``."""

    def __init__(self, status: int, body: dict[str, Any]) -> None:
        error = body.get("error", {})
        self.status = status
        self.code = error.get("code")
        super().__init__(f"HTTP {status} {self.code}: {error.get('message', body)}")


class LibrarioClient:
    def __init__(self, base_url: str, access_token: str, timeout: float = 10.0) -> None:
        self._http = httpx.Client(
            base_url=f"{base_url.rstrip('/')}/api/v3",
            headers={"Authorization": f"Bearer {access_token}", "Accept": "application/json"},
            timeout=timeout,
        )

    def __enter__(self) -> LibrarioClient:
        return self

    def __exit__(self, *_exc: object) -> None:
        self._http.close()

    def whoami(self) -> dict[str, Any]:
        """The authenticated user. The cheapest way to check that a token works."""
        return self._request("GET", "/users/me")

    def publications(self, **filters: Any) -> dict[str, Any]:
        """Filters are the flat query parameters of ``GET /publications``: doi, isbn, issn,
        arxiv_id, pmid, author, category, year, query (full text), ...

        The typed ``identifiers`` array is the *response* shape; you still search by the flat
        parameter.
        """
        return self._request("GET", "/publications", params=filters)

    def each_publication(self, per_page: int = 25, **filters: Any) -> Iterator[dict[str, Any]]:
        """Yields every publication matching the filters. List endpoints cap at 25 per page."""
        page = 1
        while True:
            records = self.publications(page=page, per_page=per_page, **filters)["records"]
            yield from records
            if len(records) < per_page:
                return
            page += 1

    def create_publication(self, attributes: dict[str, Any]) -> dict[str, Any]:
        return self._request("POST", "/publications", json=attributes)

    def publication(self, publication_id: int) -> dict[str, Any]:
        return self._request("GET", f"/publications/{publication_id}")

    def delete_publication(self, publication_id: int) -> dict[str, Any]:
        return self._request("DELETE", f"/publications/{publication_id}")

    def users(self, **filters: Any) -> dict[str, Any]:
        return self._request("GET", "/users", params=filters)

    def update_user(self, user_id: int, attributes: dict[str, Any]) -> dict[str, Any]:
        return self._request("PATCH", f"/users/{user_id}", json=attributes)

    def presign_asset(self, filename: str, content_type: str) -> dict[str, Any]:
        """Presigns a direct-to-S3 upload. ``fields["key"]`` becomes the asset's file id once the
        ``cache/`` prefix is stripped."""
        return self._request("GET", "/assets/presign", params={"filename": filename, "type": content_type})

    def presign_cover(self, filename: str, content_type: str) -> dict[str, Any]:
        """Presigns a direct-to-S3 cover-image upload. ``fields["key"]`` becomes the publication's
        cover id once the ``cache/`` prefix is stripped. Set the resulting
        ``{"id", "storage": "cache", "metadata"}`` hash as ``cover`` on a publication create/update."""
        return self._request("GET", "/publications/cover/presign", params={"filename": filename, "type": content_type})

    def create_asset(self, attributes: dict[str, Any]) -> dict[str, Any]:
        return self._request("POST", "/assets", json=attributes)

    def _request(self, method: str, path: str, **kwargs: Any) -> dict[str, Any]:
        response = self._http.request(method, path, **kwargs)
        body = response.json() if response.content else {}
        if response.is_success:
            return body
        raise LibrarioError(response.status_code, body)

Availability check

Ruby

#!/usr/bin/env ruby
# frozen_string_literal: true

require 'csv'
require_relative 'librario_client'

# Reconciles a reference-manager export against the Librario holdings.
#
# Reads a CSV with a DOI column, asks Librario about each DOI, and writes the same rows back
# with two extra columns: whether the publication is held, and where to find it.
#
#   LIBRARIO_BASE_URL=https://acme.mylibrar.io \
#   LIBRARIO_ACCESS_TOKEN=... \
#     ruby check_availability.rb endnote_export.csv availability_report.csv
#
# Needs the `library:read` scope.
class CheckAvailability
  def initialize(client:, base_url:)
    @client = client
    @base_url = base_url.to_s.chomp('/')
  end

  # @return [Array<CSV::Row>] the input rows, each with Status and URL filled in
  def call(input_path:, output_path:)
    rows = CSV.read(input_path, headers: true)
    rows.each { |row| annotate(row) }

    CSV.open(output_path, 'w') do |csv|
      csv << (rows.headers | %w[Status URL])
      rows.each { |row| csv << row }
    end
    rows
  end

  # Looks a DOI up and returns the publication, or nil.
  #
  # v3 answers with a typed `identifiers` array instead of v2's flat `doi`/`isbn` fields, but you
  # still *search* by the flat query parameter. Only the response shape changed.
  def lookup(doi)
    @client.publications(doi:, per_page: 1).fetch('records').first
  end

  private

  def annotate(row)
    doi = row['DOI'].to_s.strip
    publication = doi.empty? ? nil : lookup(doi)

    row['Status'] = publication ? 'available' : 'not available'
    row['URL'] = publication ? publication_url(publication) : ''
  end

  def publication_url(publication)
    "#{@base_url}/publications/#{publication.fetch('id')}"
  end
end

if __FILE__ == $PROGRAM_NAME
  input, output = ARGV
  abort "usage: #{$PROGRAM_NAME} <input.csv> <output.csv>" unless input && output

  base_url = ENV.fetch('LIBRARIO_BASE_URL')
  client = LibrarioClient.new(base_url:, access_token: ENV.fetch('LIBRARIO_ACCESS_TOKEN'))
  rows = CheckAvailability.new(client:, base_url:).call(input_path: input, output_path: output)
  held = rows.count { |row| row['Status'] == 'available' }
  puts "#{held} of #{rows.size} publications are held; wrote #{output}"
end

Python

#!/usr/bin/env python3
"""Reconciles a reference-manager export against the Librario holdings.

Reads a CSV with a DOI column, asks Librario about each DOI, and writes the same rows back with
two extra columns: whether the publication is held, and where to find it.

    LIBRARIO_BASE_URL=https://acme.mylibrar.io \
    LIBRARIO_ACCESS_TOKEN=... \
      python check_availability.py endnote_export.csv availability_report.csv

Needs the `library:read` scope.
"""

from __future__ import annotations

import csv
import os
import sys
from typing import Any

from librario_client import LibrarioClient


class CheckAvailability:
    def __init__(self, client: LibrarioClient, base_url: str) -> None:
        self._client = client
        self._base_url = base_url.rstrip("/")

    def lookup(self, doi: str) -> dict[str, Any] | None:
        """Looks a DOI up and returns the publication, or None.

        v3 answers with a typed ``identifiers`` array instead of v2's flat ``doi``/``isbn``
        fields, but you still *search* by the flat query parameter. Only the response shape
        changed.
        """
        records = self._client.publications(doi=doi, per_page=1)["records"]
        return records[0] if records else None

    def call(self, input_path: str, output_path: str) -> list[dict[str, str]]:
        with open(input_path, newline="", encoding="utf-8") as handle:
            rows = list(csv.DictReader(handle))

        for row in rows:
            self._annotate(row)

        fieldnames = list(rows[0].keys()) if rows else ["DOI", "Status", "URL"]
        with open(output_path, "w", newline="", encoding="utf-8") as handle:
            writer = csv.DictWriter(handle, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows(rows)
        return rows

    def _annotate(self, row: dict[str, str]) -> None:
        doi = (row.get("DOI") or "").strip()
        publication = self.lookup(doi) if doi else None

        row["Status"] = "available" if publication else "not available"
        row["URL"] = f"{self._base_url}/publications/{publication['id']}" if publication else ""


def main() -> None:
    if len(sys.argv) != 3:
        sys.exit(f"usage: {sys.argv[0]} <input.csv> <output.csv>")

    base_url = os.environ["LIBRARIO_BASE_URL"]
    with LibrarioClient(base_url=base_url, access_token=os.environ["LIBRARIO_ACCESS_TOKEN"]) as client:
        rows = CheckAvailability(client, base_url).call(sys.argv[1], sys.argv[2])

    held = sum(1 for row in rows if row["Status"] == "available")
    print(f"{held} of {len(rows)} publications are held; wrote {sys.argv[2]}")


if __name__ == "__main__":
    main()

Data import with files

Ruby

#!/usr/bin/env ruby
# frozen_string_literal: true

require 'base64'
require 'net/http'
require_relative 'librario_client'

# Imports a publication and attaches a file to it.
#
# Attaching a file is a three-step handshake, because the bytes never pass through the Librario
# API. You ask for a presigned S3 upload, PUT the bytes straight to S3, then tell Librario where
# they landed:
#
#   POST /publications        -> the record
#   GET  /assets/presign      -> a presigned S3 form
#   POST <s3 url>             -> the bytes (multipart, straight to S3)
#   POST /assets              -> links the uploaded file to the publication
#
#   LIBRARIO_BASE_URL=https://acme.mylibrar.io \
#   LIBRARIO_ACCESS_TOKEN=... \
#     ruby import_publication.rb report.pdf
#
# Needs the `library:write` scope.
class ImportPublication
  # The presigned key is prefixed with "cache/"; the asset's file id is that key WITHOUT the
  # prefix. Librario looks the upload up at "cache/#{file_id}". Get this wrong and asset creation
  # fails with a NoSuchKey error from S3 rather than anything the API can explain.
  CACHE_PREFIX = %r{\Acache/}

  # The payload the script below sends. A named constant rather than an inline literal so the
  # test suite can drive this exact hash — the version that only existed inside `__main__` drifted
  # out of shape unnoticed, because nothing could reach it.
  #
  # `categories` takes objects, not bare strings: each entry is a `{name:}` (find-or-create) or an
  # `{id:}` (associate an existing one). A bare string is rejected with 422.
  EXAMPLE_ATTRIBUTES = {
    title: 'Technische Dokumentation 2026',
    publishable_type: 'Report',
    published_on: '2026-01-15',
    authors: [{ name: 'Schmidt, Maria', type: 'person' }],
    identifiers: [{ type: 'doi', value: '10.1234/example.2026' }],
    categories: [{ name: 'Technik' }, { name: 'Forschungsbericht' }]
  }.freeze

  def initialize(client:)
    @client = client
  end

  def call(attributes, file_path: nil)
    publication = @client.create_publication(attributes)
    attach(publication_id: publication.fetch('id'), file_path:) if file_path
    publication
  end

  def attach(publication_id:, file_path:)
    presign = @client.presign_asset(filename: File.basename(file_path), type: 'application/pdf')
    upload_to_s3(presign, file_path)

    @client.create_asset(
      publication_id:,
      name: File.basename(file_path),
      content_type: 'application/pdf',
      file: {
        id: file_id(presign),
        storage: 'cache',
        metadata: {
          size: File.size(file_path),
          filename: File.basename(file_path),
          mime_type: 'application/pdf'
        }
      }
    )
  end

  # The file id Librario expects, derived from the presigned key.
  def file_id(presign) = presign.fetch('fields').fetch('key').sub(CACHE_PREFIX, '')

  private

  # S3 validates the form fields against the signed policy, so every field comes back verbatim and
  # `file` must be the last part of the multipart body.
  def upload_to_s3(presign, file_path)
    uri = URI.parse(presign.fetch('url'))
    fields = presign.fetch('fields')

    request = Net::HTTP::Post.new(uri)
    request.set_form(
      fields.map { |name, value| [name, value.to_s] } + [['file', File.open(file_path, 'rb')]],
      'multipart/form-data'
    )

    response = Net::HTTP.start(uri.hostname, uri.port, use_ssl: uri.scheme == 'https') { |http| http.request(request) }
    return if response.is_a?(Net::HTTPSuccess)

    raise LibrarioClient::Error.new(response.code.to_i, { 'error' => { 'message' => response.body.to_s[0, 200] } })
  end
end

if __FILE__ == $PROGRAM_NAME
  path = ARGV.first
  abort "usage: #{$PROGRAM_NAME} <file.pdf>" unless path

  client = LibrarioClient.new(
    base_url: ENV.fetch('LIBRARIO_BASE_URL'),
    access_token: ENV.fetch('LIBRARIO_ACCESS_TOKEN')
  )
  publication = ImportPublication.new(client:).call(ImportPublication::EXAMPLE_ATTRIBUTES, file_path: path)
  puts "imported publication #{publication.fetch('id')}"
end

Python

#!/usr/bin/env python3
"""Imports a publication and attaches a file to it.

Attaching a file is a three-step handshake, because the bytes never pass through the Librario
API. You ask for a presigned S3 upload, POST the bytes straight to S3, then tell Librario where
they landed:

    POST /publications        -> the record
    GET  /assets/presign      -> a presigned S3 form
    POST <s3 url>             -> the bytes (multipart, straight to S3)
    POST /assets              -> links the uploaded file to the publication

    LIBRARIO_BASE_URL=https://acme.mylibrar.io \
    LIBRARIO_ACCESS_TOKEN=... \
      python import_publication.py report.pdf

Needs the `library:write` scope.
"""

from __future__ import annotations

import os
import sys
from pathlib import Path
from typing import Any

import httpx

from librario_client import LibrarioClient

CONTENT_TYPE = "application/pdf"

# The payload the script below sends. A named constant rather than an inline literal so the test
# suite can drive this exact dict — the version that only existed inside ``main()`` drifted out of
# shape unnoticed, because nothing could reach it.
#
# ``categories`` takes objects, not bare strings: each entry is a ``{"name": ...}``
# (find-or-create) or an ``{"id": ...}`` (associate an existing one). A bare string is rejected
# with 422.
EXAMPLE_ATTRIBUTES: dict[str, Any] = {
    "title": "Technische Dokumentation 2026",
    "publishable_type": "Report",
    "published_on": "2026-01-15",
    "authors": [{"name": "Schmidt, Maria", "type": "person"}],
    "identifiers": [{"type": "doi", "value": "10.1234/example.2026"}],
    "categories": [{"name": "Technik"}, {"name": "Forschungsbericht"}],
}


class ImportPublication:
    def __init__(self, client: LibrarioClient) -> None:
        self._client = client

    def call(self, attributes: dict[str, Any], file_path: str | None = None) -> dict[str, Any]:
        publication = self._client.create_publication(attributes)
        if file_path:
            self.attach(publication["id"], file_path)
        return publication

    def attach(self, publication_id: int, file_path: str) -> dict[str, Any]:
        path = Path(file_path)
        presign = self._client.presign_asset(filename=path.name, content_type=CONTENT_TYPE)
        self._upload_to_s3(presign, path)

        return self._client.create_asset(
            {
                "publication_id": publication_id,
                "name": path.name,
                "content_type": CONTENT_TYPE,
                "file": {
                    "id": self.file_id(presign),
                    "storage": "cache",
                    "metadata": {
                        "size": path.stat().st_size,
                        "filename": path.name,
                        "mime_type": CONTENT_TYPE,
                    },
                },
            }
        )

    @staticmethod
    def file_id(presign: dict[str, Any]) -> str:
        """The presigned key is prefixed with ``cache/``; the asset's file id is that key WITHOUT
        the prefix. Librario looks the upload up at ``cache/{file_id}``. Get this wrong and asset
        creation fails with a NoSuchKey error from S3 rather than anything the API can explain.
        """
        return presign["fields"]["key"].removeprefix("cache/")

    @staticmethod
    def _upload_to_s3(presign: dict[str, Any], path: Path) -> None:
        """S3 validates the form fields against the signed policy, so every field goes back
        verbatim and ``file`` must be the last part of the multipart body."""
        with path.open("rb") as handle:
            response = httpx.post(
                presign["url"],
                data=presign["fields"],
                files={"file": (path.name, handle, CONTENT_TYPE)},
                timeout=30.0,
            )
        response.raise_for_status()


def main() -> None:
    if len(sys.argv) != 2:
        sys.exit(f"usage: {sys.argv[0]} <file.pdf>")

    with LibrarioClient(
        base_url=os.environ["LIBRARIO_BASE_URL"],
        access_token=os.environ["LIBRARIO_ACCESS_TOKEN"],
    ) as client:
        publication = ImportPublication(client).call(EXAMPLE_ATTRIBUTES, file_path=sys.argv[1])
    print(f"imported publication {publication['id']}")


if __name__ == "__main__":
    main()