--- type: Documentation title: REST API description: Die Librario REST API ermöglicht die programmatische Integration und Automatisierung Ihrer Bibliotheksprozesse. resource: https://www.librario.de/de/docs/integration/api tags: - docs - integration --- # REST API > **Tarif-Option:** > > Verfügbar in Tarifen mit **REST API**. [Tarife vergleichen→](/de/preise) > **Kein Web Scraping:** > > Bitte verwenden Sie für automatisierte Zugriffe ausschließlich die REST API. Das Scraping der Weboberfläche wird nicht unterstützt und kann zu unerwarteten Fehlern führen. Die API bietet alle notwendigen Endpunkte für eine programmatische Integration. 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** | **V2** | | --- | --- | --- | | Status | Preview | Veraltet, Abschaltung zum Jahresende 2026 | | Basis-URL | `/api/v3` | `/api/v2` | | Anmeldung | OAuth 2.1 | Basic Authentication oder Session-Cookie | | Identifikatoren | typisierte `identifiers`-Liste | flache Felder `doi`, `isbn`, `pmid`, … | 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](/de/changelog/2026-03-api-v3-preview). > **V3 ist eine Preview:** > > Die Schnittstelle kann sich bis zur stabilen Freigabe noch ändern. Für den Produktivbetrieb empfehlen wir, die Migration jetzt zu planen und erst nach dem stabilen V3-Release umzusetzen. V2 bleibt bis zur Abschaltung unverändert verfügbar. ## Zugriff auf die API Die API ist unter folgender URL erreichbar: `https://.mylibrar.io/api`, zum Beispiel: Falls Sie eine eigene Subdomain zugewiesen bekommen haben, können Sie diese in der URL verwenden. ## API-Dokumentation ![Screenshot der Swagger UI Dokumentation der Librario API. Links sind die verschiedenen API-Endpunkte nach Ressourcen gruppiert zu sehen. Rechts werden die Details des ausgewählten Endpunkts mit Parametern und Beispielantworten angezeigt.](/de/docs/integration/api/swagger-ui.webp) _Die interaktive API-Dokumentation mittels Swagger UI_ 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://.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://.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. > **Zum Ausprobieren genügt ein Klick:** > > Öffnen Sie `https://.mylibrar.io/api` und klicken Sie auf **Authorize**. Client, PKCE und Zielressource sind dort bereits hinterlegt. Nach der Anmeldung können Sie jeden Endpunkt direkt in der Swagger UI aufrufen. 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 ```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 ```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 ``` > **Client-ID aufbewahren:** > > Die Registrierung ist einmalig. Speichern Sie die zurückgegebene `client_id` und geben Sie sie beim nächsten Lauf mit, statt sich erneut zu registrieren. Jede Bibliothek nimmt pro Tag nur eine begrenzte Zahl an Registrierungen entgegen. ### V3: Berechtigungen Ein Token trägt die Schnittmenge aus den angefragten Berechtigungen und dem, was die anmeldende Person auf der Zustimmungsseite freigegeben hat. | Scope | Erlaubt | | --- | --- | | `library:read` | Publikationen, Dateien, Exemplare und Normdaten lesen | | `library:write` | Bibliotheksdaten anlegen, ändern und löschen | | `users:read` | Benutzerverzeichnis lesen | | `users:write` | Benutzer:innen verwalten | | `tasks:read` | Aufgaben und Qualitätsprüfungen lesen | | `tasks:write` | Aufgaben anlegen, bearbeiten und abschließen | 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](/de/docs/administration/rollen). > **Technische Benutzer für Integrationen:** > > Es wird empfohlen, für automatisierte Prozesse dedizierte technische Zugänge anzulegen. Dies macht die Integration unabhängig von personenbezogenen Accounts und vereinfacht die Verwaltung von Zugriffsrechten. Beispiele für V2 führen wir nicht mehr auf. Die vollständige Beschreibung finden Sie in der OpenAPI-Spezifikation unter `https://.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 ```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 "/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 ```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 ``/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) ``` ## 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. > **Suche und Antwort verwenden unterschiedliche Formen:** > > V3 liefert Identifikatoren als typisierte `identifiers`-Liste zurück. Gesucht wird weiterhin über den flachen Parameter `doi`, `isbn`, `issn`, `arxiv_id` oder `pmid`. #### Ruby ```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] 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} " 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 ```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]} ") 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. ![Sequenzdiagramm des Datei-Uploads: Der Client erstellt über die API eine Publikation, fordert eine presigned URL an, lädt die Datei per Multipart-Upload direkt zu S3 hoch, extrahiert die File-ID und legt damit das Asset über die API an.](/de/docs/integration/api/upload-sequenz.svg) > **Wichtiger Hinweis zur File ID:** > > Die `file.id` für die Asset-Erstellung entspricht dem S3-Pfad **ohne** das `cache/` Präfix. > > **Beispiel:** > > ``` > S3 Key: cache/1/a078705e6e55b512bb2b596c12a2a12b/dokument.pdf > File ID: 1/a078705e6e55b512bb2b596c12a2a12b/dokument.pdf > ``` > > Das Backend sucht die Datei unter `cache/{file.id}` in S3. Bei falscher ID erhält man den Fehler: > > ``` > file "..." not found on storage (Aws::S3::Errors::NoSuchKey) > ``` #### Ruby ```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 -\> 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} " 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 ```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 -\> 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]} ") 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() ``` ### Backups und Datensicherung Die API bietet umfassende Möglichkeiten zur [Datensicherung](/de/docs/administration/backups). Ein einfaches Backup mittels `/publications/all` Endpunkt ermöglicht das seitenweise Abrufen aller Publikationen. Weitere Details finden Sie in der [Backup-Dokumentation](/de/docs/administration/backups). ## 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: ```json { "error": { "code": "RecordNotFound", "message": "Requested record not found" } } ```