REST API

Die Librario REST API ermöglicht Ihnen die programmatische Integration und Automatisierung Ihrer Bibliotheksprozesse. Sie können damit alle wichtigen Funktionen von Librario in Ihre bestehende IT-Infrastruktur bzw. in Ihre Prozesse einbinden.

Zwei API-Versionen

Es gibt zurzeit zwei Versionen der API. Sie sind unabhängig voneinander erreichbar und unterscheiden sich sowohl im Datenmodell als auch in der Anmeldung.

V3 räumt Altlasten der V2 auf: eigene Endpunkte für verschachtelte Ressourcen, dedizierte DELETE-Endpunkte statt _destroy-Markern und ein vereinheitlichtes PATCH /users/:id. Die Details stehen im Changelog-Eintrag API V3 als Preview verfügbar.

Zugriff auf die API

Die API ist unter folgender URL erreichbar:

https://<ihre-firma>.mylibrar.io/api, zum Beispiel: https://app.mylibrar.io/api

Falls Sie eine eigene Subdomain zugewiesen bekommen haben, können Sie diese in der URL verwenden.

API-Dokumentation

Die vollständige API-Dokumentation steht Ihnen in zwei Formaten zur Verfügung:

1. Swagger UI: Eine interaktive Web-Oberfläche zum Erkunden der API

   • Erreichbar unter https://<ihre-firma>.mylibrar.io/api
   • Ermöglicht das direkte Testen von API-Aufrufen
   • Zeigt Beispielanfragen und -antworten
   • Bietet Download der OpenAPI-Spezifikation als swagger.yaml

2. OpenAPI-Spezifikation (swagger.yaml): Eine maschinenlesbare Beschreibung der API

   • Basis für die automatische Generierung von API-Clients
   • Kann in API-Entwicklungswerkzeuge importiert werden
   • Hilfreich für die Verwendung mit KI-Tools wie ChatGPT oder Claude

Authentifizierung

Die beiden API-Versionen melden sich unterschiedlich an.

V3: OAuth 2.1

V3 akzeptiert ausschließlich OAuth 2.1 mit dem Authorization-Code-Flow und PKCE (Verfahren S256). Es gibt keine Client-Secrets und keine Anmeldung mit Benutzername und Passwort. Ein Zugriffstoken ist eine Stunde gültig und bringt ein Refresh-Token mit, das sich bei jeder Verwendung erneuert.

Jedes Token gilt für genau eine Zielressource, nämlich https://<ihre-firma>.mylibrar.io/api/v3. Diese Adresse geben Sie im Parameter resource an, sowohl beim Autorisieren als auch beim Einlösen des Codes. Fehlt der Parameter, antwortet der Autorisierungsserver mit error=invalid_target, denn ein Token ohne festgelegte Zielressource ließe sich gegen eine andere Schnittstelle wiederverwenden.

Für Skripte übernimmt das folgende Beispiel den vollständigen Ablauf: Es registriert sich selbst als OAuth-Client (Dynamic Client Registration), erzeugt den PKCE-Nachweis, nimmt die Weiterleitung auf einem lokalen Port entgegen und tauscht den Code gegen ein Token.

Ruby

# frozen_string_literal: true

require 'base64'
require 'digest'
require 'json'
require 'net/http'
require 'securerandom'
require 'socket'
require 'uri'

# Obtains a Librario v3 access token from a script, with no client secret anywhere.
#
# The whole OAuth 2.1 dance, in four steps:
#
#   1. Register this script as an OAuth client (RFC 7591). No pre-arrangement with Librario,
#      no secret. Keep the returned client_id; re-registering on every run is wasteful and
#      each Librario tenant only accepts a bounded number of registrations per day.
#   2. Generate a PKCE verifier and its S256 challenge (RFC 7636). Librario accepts S256 only.
#   3. Send the user to /oauth/authorize and catch the redirect on a loopback listener
#      (RFC 8252). Loopback is the one place Librario permits a plain-http callback.
#   4. Exchange the code for a token, proving possession of the verifier.
#
# `resource` (RFC 8707) appears in steps 3 and 4. It names which Librario surface the token is
# for. Omit it and the authorization server refuses with `invalid_target`, because a token has
# to be bound to exactly one audience.
#
#   token = OAuthPkceLogin.new(base_url: 'https://acme.mylibrar.io').call
#   client = LibrarioClient.new(base_url: 'https://acme.mylibrar.io', access_token: token['access_token'])
#
# Access tokens live one hour. `token['refresh_token']` renews them; it rotates on every use.
class OAuthPkceLogin
  class Error < StandardError
  end

  DEFAULT_SCOPE = 'library:read library:write users:read users:write tasks:read tasks:write'

  # @param base_url [String] e.g. "https://acme.mylibrar.io"
  # @param client_id [String, nil] from a previous run; registers a new client when nil
  # @param scope [String] space-delimited
  # @param port [Integer] loopback port for the callback listener
  def initialize(base_url:, client_id: nil, scope: DEFAULT_SCOPE, port: 8765)
    @base_url = base_url.to_s.chomp('/')
    @client_id = client_id
    @scope = scope
    @port = port
  end

  # Runs the full flow and returns the parsed token response.
  #
  # Yields the authorization URL when a block is given, instead of opening a browser. That is
  # what the test suite uses; a real script wants the default.
  def call(&open_url)
    client_id = @client_id || register_client.fetch('client_id')
    verifier = generate_code_verifier
    state = SecureRandom.urlsafe_base64(16)

    server = TCPServer.new('127.0.0.1', @port)
    url = authorization_url(client_id:, verifier:, state:)
    open_url ? yield(url) : open_browser(url)

    code = await_callback(server, expected_state: state)
    exchange_code(code:, client_id:, verifier:)
  ensure
    server&.close
  end

  # RFC 7591. The redirect URI must be a loopback IP with an explicit port: Librario rejects
  # `http://localhost/...` without one, because Doorkeeper's any-port loopback matching only
  # applies to IP literals and such a client could never complete a login.
  def register_client
    post_form(
      "#{@base_url}/oauth/clients",
      client_name: 'Librario example script',
      redirect_uris: [redirect_uri],
      grant_types: %w[authorization_code refresh_token],
      response_types: %w[code],
      token_endpoint_auth_method: 'none',
      scope: @scope
    )
  end

  def authorization_url(client_id:, verifier:, state:)
    query = {
      response_type: 'code',
      client_id:,
      redirect_uri:,
      scope: @scope,
      state:,
      code_challenge: code_challenge(verifier),
      code_challenge_method: 'S256',
      resource: api_v3_audience
    }
    "#{@base_url}/oauth/authorize?#{URI.encode_www_form(query)}"
  end

  def exchange_code(code:, client_id:, verifier:)
    post_form(
      "#{@base_url}/oauth/token",
      grant_type: 'authorization_code',
      code:,
      redirect_uri:,
      client_id:,
      code_verifier: verifier,
      resource: api_v3_audience
    )
  end

  private

  def post_form(url, **params)
    uri = URI.parse(url)
    request = Net::HTTP::Post.new(uri)
    request['Content-Type'] = 'application/json'
    request['Accept'] = 'application/json'
    request.body = JSON.generate(params)

    response = Net::HTTP.start(uri.hostname, uri.port, use_ssl: uri.scheme == 'https') { |http| http.request(request) }
    body = response.body.to_s.empty? ? {} : JSON.parse(response.body)
    raise Error, "#{uri.path} failed: HTTP #{response.code} #{body}" unless response.is_a?(Net::HTTPSuccess)

    body
  end

  def redirect_uri = "http://127.0.0.1:#{@port}/callback"
  def generate_code_verifier = SecureRandom.urlsafe_base64(32)

  def code_challenge(verifier)
    Base64.urlsafe_encode64(Digest::SHA256.digest(verifier), padding: false)
  end

  def api_v3_audience = "#{@base_url}/api/v3"

  def open_browser(url)
    opener = %w[xdg-open open].find { |cmd| system('which', cmd, out: File::NULL, err: File::NULL) }
    opener ? system(opener, url, out: File::NULL, err: File::NULL) : warn("Open this URL to authorize:\n\n  #{url}\n")
  end

  # Blocks until the browser follows the redirect back to the loopback listener.
  def await_callback(server, expected_state:)
    socket = server.accept
    query = callback_query(socket.gets.to_s)

    respond(socket, query['error'] ? "Authorization failed: #{query['error']}" : 'Authorized. You can close this tab.')
    socket.close

    raise Error, "authorization failed: #{query['error']}" if query['error']
    raise Error, 'state mismatch — possible CSRF, aborting' unless query['state'] == expected_state

    query.fetch('code')
  end

  # "GET /callback?code=...&state=... HTTP/1.1" -> { "code" => ..., "state" => ... }
  def callback_query(request_line)
    path = request_line.split[1].to_s
    URI.decode_www_form(URI.parse(path).query.to_s).to_h
  end

  def respond(socket, message)
    socket.print("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: #{message.bytesize}\r\n\r\n#{message}")
  end
end

Python

"""Obtains a Librario v3 access token from a script, with no client secret anywhere.

The whole OAuth 2.1 dance, in four steps:

1. Register this script as an OAuth client (RFC 7591). No pre-arrangement with Librario, no
   secret. Keep the returned ``client_id``: re-registering on every run is wasteful, and each
   Librario tenant only accepts a bounded number of registrations per day.
2. Generate a PKCE verifier and its S256 challenge (RFC 7636). Librario accepts S256 only.
3. Send the user to ``/oauth/authorize`` and catch the redirect on a loopback listener (RFC 8252).
   Loopback is the one place Librario permits a plain-http callback.
4. Exchange the code for a token, proving possession of the verifier.

``resource`` (RFC 8707) appears in steps 3 and 4. It names which Librario surface the token is
for. Omit it and the authorization server refuses with ``invalid_target``, because a token has to
be bound to exactly one audience.

    token = OAuthPkceLogin(base_url="https://acme.mylibrar.io").call()
    client = LibrarioClient(base_url="https://acme.mylibrar.io", access_token=token["access_token"])

Access tokens live one hour. ``token["refresh_token"]`` renews them; it rotates on every use.
"""

from __future__ import annotations

import base64
import hashlib
import http.server
import secrets
import threading
import urllib.parse
import webbrowser
from collections.abc import Callable
from typing import Any

import httpx

DEFAULT_SCOPE = "library:read library:write users:read users:write tasks:read tasks:write"


class OAuthError(Exception):
    pass


class OAuthPkceLogin:
    def __init__(
        self,
        base_url: str,
        client_id: str | None = None,
        scope: str = DEFAULT_SCOPE,
        port: int = 8765,
    ) -> None:
        self.base_url = base_url.rstrip("/")
        self.client_id = client_id
        self.scope = scope
        self.port = port

    @property
    def redirect_uri(self) -> str:
        return f"http://127.0.0.1:{self.port}/callback"

    @property
    def api_v3_audience(self) -> str:
        return f"{self.base_url}/api/v3"

    def call(self, open_url: Callable[[str], None] | None = None) -> dict[str, Any]:
        """Runs the full flow and returns the parsed token response.

        ``open_url`` receives the authorization URL instead of a browser being opened. That is
        what the test suite uses; a real script wants the default.
        """
        client_id = self.client_id or self.register_client()["client_id"]
        verifier = secrets.token_urlsafe(32)
        state = secrets.token_urlsafe(16)

        with _CallbackListener(self.port) as listener:
            url = self.authorization_url(client_id=client_id, verifier=verifier, state=state)
            (open_url or webbrowser.open)(url)
            query = listener.wait()

        if "error" in query:
            raise OAuthError(f"authorization failed: {query['error']}")
        if query.get("state") != state:
            raise OAuthError("state mismatch — possible CSRF, aborting")

        return self.exchange_code(code=query["code"], client_id=client_id, verifier=verifier)

    def register_client(self) -> dict[str, Any]:
        """RFC 7591. The redirect URI must be a loopback IP with an explicit port: Librario
        rejects ``http://localhost/...`` without one, because Doorkeeper's any-port loopback
        matching only applies to IP literals and such a client could never complete a login.
        """
        return self._post_json(
            f"{self.base_url}/oauth/clients",
            {
                "client_name": "Librario example script",
                "redirect_uris": [self.redirect_uri],
                "grant_types": ["authorization_code", "refresh_token"],
                "response_types": ["code"],
                "token_endpoint_auth_method": "none",
                "scope": self.scope,
            },
        )

    def authorization_url(self, client_id: str, verifier: str, state: str) -> str:
        query = urllib.parse.urlencode(
            {
                "response_type": "code",
                "client_id": client_id,
                "redirect_uri": self.redirect_uri,
                "scope": self.scope,
                "state": state,
                "code_challenge": _code_challenge(verifier),
                "code_challenge_method": "S256",
                "resource": self.api_v3_audience,
            }
        )
        return f"{self.base_url}/oauth/authorize?{query}"

    def exchange_code(self, code: str, client_id: str, verifier: str) -> dict[str, Any]:
        return self._post_json(
            f"{self.base_url}/oauth/token",
            {
                "grant_type": "authorization_code",
                "code": code,
                "redirect_uri": self.redirect_uri,
                "client_id": client_id,
                "code_verifier": verifier,
                "resource": self.api_v3_audience,
            },
        )

    @staticmethod
    def _post_json(url: str, payload: dict[str, Any]) -> dict[str, Any]:
        response = httpx.post(url, json=payload, headers={"Accept": "application/json"}, timeout=10.0)
        body = response.json() if response.content else {}
        if not response.is_success:
            raise OAuthError(f"{urllib.parse.urlparse(url).path} failed: HTTP {response.status_code} {body}")
        return body


def _code_challenge(verifier: str) -> str:
    digest = hashlib.sha256(verifier.encode()).digest()
    return base64.urlsafe_b64encode(digest).rstrip(b"=").decode()


class _CallbackListener:
    """Serves exactly one request on the loopback port and hands back its query string."""

    def __init__(self, port: int) -> None:
        self._query: dict[str, str] = {}
        listener = self

        class Handler(http.server.BaseHTTPRequestHandler):
            def do_GET(self) -> None:  # noqa: N802 - name fixed by BaseHTTPRequestHandler
                parsed = urllib.parse.urlparse(self.path)
                listener._query = dict(urllib.parse.parse_qsl(parsed.query))
                body = b"Authorized. You can close this tab."
                self.send_response(200)
                self.send_header("Content-Type", "text/plain")
                self.send_header("Content-Length", str(len(body)))
                self.end_headers()
                self.wfile.write(body)

            def log_message(self, *_args: Any) -> None:
                pass

        self._server = http.server.HTTPServer(("127.0.0.1", port), Handler)
        self._thread = threading.Thread(target=self._server.handle_request, daemon=True)

    def __enter__(self) -> _CallbackListener:
        self._thread.start()
        return self

    def __exit__(self, *_exc: object) -> None:
        self._server.server_close()

    def wait(self, timeout: float = 300.0) -> dict[str, str]:
        self._thread.join(timeout)
        if self._thread.is_alive():
            raise OAuthError("timed out waiting for the authorization redirect")
        return self._query

V3: Berechtigungen

Ein Token trägt die Schnittmenge aus den angefragten Berechtigungen und dem, was die anmeldende Person auf der Zustimmungsseite freigegeben hat.

Fehlt einem Token eine Berechtigung, antwortet die API mit 403 InsufficientScope und nicht mit 401.

V2: Basic Authentication oder Session-Cookie

V2 kennt zwei Verfahren:

1. Basic Authentication mit den normalen Librario-Zugangsdaten.
2. Session-Cookie (_library_session), wenn der Aufruf aus einer bereits angemeldeten Browser-Sitzung erfolgt.

Die Berechtigungen entsprechen in beiden Fällen den in der Anwendung konfigurierten Rollen.

Beispiele für V2 führen wir nicht mehr auf. Die vollständige Beschreibung finden Sie in der OpenAPI-Spezifikation unter https://<ihre-firma>.mylibrar.io/api/v2/swagger.yaml.

Datenformat

Die API nutzt das JSON-Format für den Datenaustausch. Dieses Format ist in allen gängigen Programmiersprachen einfach zu verarbeiten.

Paginierung

Bei Listen-Endpoints wird die Ausgabe auf 25 Einträge pro Seite begrenzt, maximal sind 50 möglich. Um alle Datensätze zu erhalten, rufen Sie die Seiten nacheinander ab.

Die Navigation wird durch spezielle Response-Header erleichtert:

• X-Total: Gesamtanzahl der verfügbaren Einträge
• Link: URLs für erste/vorherige/nächste/letzte Seite

Ein minimaler Client

Die folgenden Beispiele bauen alle auf diesem Client auf. Er kommt ohne zusätzliche Abhängigkeiten aus und zeigt Anmeldung, Paginierung und Fehlerbehandlung an einer Stelle.

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:)
  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 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)

Typische Anwendungsfälle

Abgleich von Publikationslisten

Ein häufiges Szenario ist der Abgleich eines Literaturverwaltungs-Exports mit dem Librario-Bestand. Dabei soll ermittelt werden, welche Publikationen bereits verfügbar sind und welche noch beschafft werden müssen.

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()

Datenimport mit Dateien

Beim Umstieg auf Librario müssen oft Daten aus dem Altsystem migriert werden. Die Datei-Bytes laufen dabei nicht durch die API, sondern gehen direkt zu S3.

S3 Key: cache/1/a078705e6e55b512bb2b596c12a2a12b/dokument.pdf
File ID: 1/a078705e6e55b512bb2b596c12a2a12b/dokument.pdf

file "..." not found on storage (Aws::S3::Errors::NoSuchKey)

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/}

  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(
    {
      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: %w[Technik Forschungsbericht]
    },
    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"


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(
            {
                "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": ["Technik", "Forschungsbericht"],
            },
            file_path=sys.argv[1],
        )
    print(f"imported publication {publication['id']}")


if __name__ == "__main__":
    main()

Backups und Datensicherung

Die API bietet umfassende Möglichkeiten zur Datensicherung. Ein einfaches Backup mittels /publications/all Endpunkt ermöglicht das seitenweise Abrufen aller Publikationen. Weitere Details finden Sie in der Backup-Dokumentation.

Best Practices

Anfragerate

Bitte begrenzen Sie automatisierte Zugriffe auf etwa 60 Anfragen pro Minute, verwenden Sie bei Listenabfragen die Paginierung und verarbeiten Sie große Datenmengen in kleineren Batches.

Fehlerbehandlung

Die API verwendet folgende HTTP-Statuscodes:

• 200: Anfrage erfolgreich
• 401: Nicht autorisiert, etwa weil das Token abgelaufen ist
• 403: Zugriff verweigert oder Berechtigung fehlt (InsufficientScope)
• 404: Ressource nicht gefunden
• 422: Eingabe konnte nicht verarbeitet werden
• 429: Zu viele Anfragen
• 500: Interner Serverfehler

Fehlerantworten enthalten zusätzliche Details im JSON-Format:

{
  "error": {
    "code": "RecordNotFound",
    "message": "Requested record not found"
  }
}