The following examples assume the client variable is an instance of XataClient with the correct credentials. To learn more about initializing the SDK, refer to the Configuration page.

Each endpoint returns an instance of requests.Response. If your query reaches the threshold of concurrent connections, the server will respond with a 429 status code and the SDK will throw a RateLimitingException.

You can refer to the API Reference for the relevant parameters to each method.

More examples are available in the GitHub repository of the SDK, including runnable demo apps for the BulkProcessor, pagination of query results, and leveraging transactions.

Below are the parameters passed to the methods in the following examples:

  • xata.records().insert(): table name (str), record (dictionary)
  • xata.records().insert_with_id(): table name (str), record id (str), record (dictionary)
  • xata.records().update(): table name (str), record id (str), record (dictionary)
  • xata.records().upsert(): table name (str), record id (str), record (dictionary)
  • xata.records().get(): table name (str), record id (str)
  • xata.data().query(): table name (str), object with "columns", "filter", and "sort"
  • xata.records().delete(): table name (str), record id (str)
  • xata.records().bulk_insert(): table name (str), records (dictionary with records array)
  • xata.users().get()
  • xata.data().ask(): table name (str), question (str)
  • xata.data().ask_follow_up(): table name (str), session id (str), question (str)
  • to_rfc3339(): dt (datetime), tz (timezone, default: utc)
from xata.client import XataClient
xata = XataClient(db_name="my_db")
record = {
"name": "Peter Parker",
"job": "Spiderman",
# Insert record to table "Avengers" and let Xata generate a record Id
resp = xata.records().insert("Avengers", record)
assert resp.is_success()
print("Record Id: %s" % resp["id"])
# Insert record to table "Avengers" with your own record Id "spidey-1"
resp = xata.records().insert_with_id("Avengers", "spidey-1", record)
assert resp.is_success()
# Update the record with Id "spidey-1" in table "Avengers"
record["job"] = "your friendly neighborhood spider man"
resp = xata.records().update("Avengers", "spidey-1", record)
assert resp.is_success()
# Upsert: Update or insert a record
record = {
"name": "Bruce Banner",
"job": "Hulk",
resp = xata.records().upsert("Avengers", "hulk-1", record)
# On insert status code = 201
assert resp.status_code == 201
assert resp.is_success()
record["job"] = "the incredible hulk"
resp = xata.records().upsert("Avengers", "spidey-1", record)
# On update status code = 200
assert resp.status_code == 200
assert resp.is_success()

The following example shows how to retrieve a record with the id spidey, from the table Avengers and how to handle records that do not exist.

from xata.client import XataClient
records = XataClient().records()
spiderman = records.get("Avengers", "spidey")
print(spiderman.is_success()) # True
# {"id": "spidey", "name": "Peter Parker", "job": "spiderman"}
# If the record with the Id does not exist, the status code will be 404
batman = records.get("Avengers", "bruce-wayne")
print(batman.status_code) # 404
print(batman.is_success()) # False

The following example shows how to query a table and apply filters and sorts. We will query the table Avengers, and apply some filters.

from xata.client import XataClient
resp = xata.data().query("Avengers", {
"columns": ["name", "thumbnail"], # the columns we want returned
"filter": { "job": "spiderman" }, # optional filters to apply
"sort": { "name": "desc" } # optional sorting key and order (asc/desc)
assert resp.is_success()
# [{"id": "spidey", "name": "Peter Parker", "job": "spiderman"}]
# Note it will be an array, even if there is only one record matching the filters

The following example will show you how to paginate through your data, and introduce you to two new convenience methods, response.has_more_results() and response.get_cursor() introduced in 1.x. We want to limit the records per call to 25, and assume the people table has a total amount of records is > 25. So we need to make multiple calls to page through our data.

from xata.client import XataClient
records = xata.data().query("people", {
"page": {
"size": 25 # limit result set to 25 records
# do something with the data ..
# are more pages available ? If yes keep looping
# through until we reached the last page
while records.has_more_results():
# fetch the next page ...
records = self.xata.data().query("people", {
"page": {
"after": records.get_cursor() # get the next cursor
# do something with the data ..
from xata.client import XataClient
records = XataClient().records()
record = records.delete("Avengers", "captain-america")
print(record.status_code) # 204

The Bulk Processor is a helpful tool for performing bulk operations using the Xata Python client. It abstracts away the complexity of managing bulk operations, making it easier to perform large-scale insert operations. Using the Bulk Processor is recommended for bulk operations in Xata.

from xata.client import XataClient
xata = XataClient(api_key="REDACTED_API_KEY", db_name="my_db", branch_name="feature-042")
avengers = [
{"name": "Peter Parker", "job": "Spiderman"},
{"name": "Bruce Banner", "job": "Hulk"},
{"name": "Steve Rodgers Parker", "job": "Captain America"},
{"name": "Tony Stark", "job": "Iron Man"},
resp = xata.records().bulk_insert("Avengers", {"records": avengers})
assert resp.is_success()

In this scenario we are uploading an on disk hosted file into a table Photos through the records API. For more examples please consult the file attachment examples.

xata = XataClient()
upload = xata.records().insert("Photos", {
"title": "My new uploaded photo",
"photo": {
"name": "Nordkette, Innsbruck",
"mediaType": "image/jpeg",
"base64Content": "Tm9yZGtldHRlLCBJbm5zYnJ1Y2ssIFR5cm9sLCBBdXN0cmlhLCBMb3JlbSBJcHN1bSBsb3Jv",
"enablePublicUrl": True
assert upload.is_success()

The following snippets shows you a very simple way to read a file from disk and encode it to base 64:

file_name = "photos/nordkette.jpg"
with open(file_name, "rb") as f:
  file_content = f.read()
b64 = base64.b64encode(file_content).decode("ascii")

The ask endpoint uses search or similarity search algorithms to find relevant information from your database. Please refer to this page to get more information about the endpoint.

xata = XataClient()
answer = xata.data().ask("xata", "does xata have a python sdk")
assert answer.is_success()
print("Answer : %s" % answer["answer"])
print("Records: %s" % answer["records"])
print("Session: %s" % answer["sessionId"])
# Ask a follow up question using the sessionId from the previous question
follow_up = xata.data().ask_follow_up("xata", answer["sessionId"], "what is the best way to do bulk?")
assert follow_up.is_success()
print("Answer : %s" % follow_up["answer"])
print("Records: %s" % follow_up["records"])
from xata.client import XataClient
xata = XataClient()
user = xata.users().get()
print(user.status_code) # 200

To use the datetime data type in Xata, you must provide an RFC 3339 compliant string.

The Xata SDK provides a convenient to_rfc3339() helper function to simplify the submission of Python native datetime values. You can pass the datetime object as a parameter to the helper function, and it will convert the value to a RFC 3339 compliant string.

To specify a timezone, you can do so by using the optional timezone argument. If no timezone is specified, UTC time is applied by default.

The to_rfc3339() helper function was introduced with the v0.9.0 release.

# import the helper function
from xata.helpers import to_rfc339
from datetime import datetime
my_date = datetime.strptime("2023-03-20 13:42:00", "%Y-%m-%d %H:%M:%S")
> "2023-03-20T13:42:00+00:00"
# with time
date_without_time = datetime.strptime("2023-03-20", "%Y-%m-%d")
> "2023-03-20T00:00:00+00:00"
# With a timezone
from pytz import timezone
date_with_tz = datetime.strptime("2023-03-20 13:42:16", "%Y-%m-%d %H:%M:%S")
tz_europe_vienna = timezone("Europe/Vienna")
print(to_rfc339(date_with_tz, tz_europe_vienna))
> "2023-03-20T13:42:16+01:05"