Running statements#
A Cypher statement, a SQL statement and a statement that mixes the two all run the same way, on the same connection, in the same transaction. Four methods run them, and each returns one kind of thing:
Method |
Returns |
Use it for |
|---|---|---|
|
a |
most statements |
|
an iterator that keeps the rows on the server |
a result too large to hold |
|
a psycopg cursor |
anything psycopg offers that a |
|
a |
many small statements at once |
execute_query()#
>>> result = conn.execute_query(
... "MATCH (p:person)-[:directed]->(m:movie) WHERE p.name = %s"
... " RETURN m.title AS title, m.year AS year ORDER BY year",
... ("James Cameron",),
... )
>>> result.keys
['title', 'year']
>>> result.records
[('The Terminator', 1984), ('Terminator 2: Judgment Day', 1991), ('Avatar', 2009)]
It sends the statement, reads the whole result, and returns a Result:
|
the rows, a |
|
the column names |
|
what the statement wrote, when asked for with |
|
the PostgreSQL type of each column, as the server described it |
A Result is itself a named tuple of those four fields, so len(result) is 4 and iterating over
it gives the fields, not the rows. Iterate over result.records.
A column without an alias is named by the server: RETURN m.title is called title, and
RETURN count(*) is called count. Give a column an alias with AS when the name matters.
A query that matches nothing returns a Result with no records, but with its keys and oids,
because the columns are still described. A statement with no result at all, such as a CREATE
without RETURN, has no keys and no oids either.
Reserved arguments#
After the statement and its parameters, execute_query() takes four keyword arguments. Their names
end in an underscore so they cannot collide with anything of yours:
|
ask for the server’s composite rendering of graph values; see Graph values |
|
read what the statement wrote; see What a write changed |
|
|
|
a psycopg row factory, to have rows built as something other than tuples |
row_ takes any of psycopg’s row factories:
>>> from psycopg.rows import dict_row
>>> conn.execute_query(
... "MATCH (m:movie) WHERE m.year = 2003 RETURN m.title AS title ORDER BY title", row_=dict_row
... ).records
[{'title': 'The Matrix Reloaded'}, {'title': 'The Matrix Revolutions'}]
There is no argument for the graph, because selecting one is a statement that belongs on the connection, once; and there is none for a timeout, because setting and restoring a limit costs two more statements each time. Deadlines bound many statements for that price instead.
Cypher and SQL together#
A Cypher query can sit wherever SQL takes a subquery, and the two kinds of statement share the connection’s transaction. A graph read and a relational join can therefore be one statement:
>>> conn.execute_query("""
... SELECT m.title #>> '{}' AS title, avg(r.score) AS score
... FROM (MATCH (:person {name: 'Christopher Nolan'})-[:directed]->(m:movie)
... RETURN m.title AS title) AS m
... JOIN rating r ON r.title = m.title #>> '{}'
... GROUP BY 1 ORDER BY 1
... """).records
[('Inception', Decimal('8.9000000000000000')), ('Interstellar', Decimal('8.6500000000000000'))]
A property returned by Cypher is jsonb, so the SQL around it unwraps it with #>> '{}' to get
text. The AgensGraph manual’s Combining SQL and
Cypher
covers the rules at that boundary.
execute() and cursors#
execute() is psycopg’s own method and returns a psycopg cursor, with the graph types readable on
it:
>>> cursor = conn.execute("MATCH (m:movie) RETURN m ORDER BY m.year LIMIT 2")
>>> cursor.fetchone()
(Vertex(movie[3.8]),)
>>> cursor.fetchone()
(Vertex(movie[3.15]),)
Use it, or conn.cursor(), for what a Result does not offer: fetching a row at a time,
executemany(), COPY, or a named server-side cursor. Everything that passes through a cursor of
this connection gets the same treatment as execute_query(): the same parameter rules, the same
error handling and the same query logging.
Streaming a large result#
execute_query() holds the whole result in memory. For a result too large for that, stream()
reads it a chunk at a time through a server-side cursor, so the rows wait on the server until you
reach them:
>>> with conn.transaction():
... for (title,) in conn.stream("MATCH (m:movie) RETURN m.title ORDER BY m.title", size=5):
... if title.startswith("Blade"):
... print(title)
Blade Runner
Blade Runner 2049
size is how many rows are fetched per round trip, 100 by default. The rows are the same values
execute_query() would give, and binary_=True works the same way.
It needs a transaction, because a server-side cursor lives inside one. That is also what makes
it safe to stop early: leaving the transaction closes the cursor. On a connection in autocommit,
open one with conn.transaction(); outside one, stream() raises
agensgraph.errors.NoEnclosingTransaction. Two streams can be open at once; each names its own
cursor.
It reads, and only reads. The grammar has no way to declare a cursor over Cypher, so the
statement is wrapped as a subquery, SELECT * FROM (...), and a subquery cannot write. A statement
with CREATE, MERGE, SET, DELETE or REMOVE is refused, naming the clause, before anything
is sent. A trailing ORDER BY or LIMIT is fine; one in the middle of the statement is not, and
the server says so.
The checks happen when you start iterating, not when you call stream(), because it is a
generator.
In awaiting code it is an asynchronous iterator, used with async for, not awaited:
async with aconn.transaction():
async for (title,) in aconn.stream("MATCH (m:movie) RETURN m.title"):
...
DataFrames and Arrow turns a stream into Arrow record batches as it goes.
What a write changed#
Pass counts_=True to learn what a statement wrote:
>>> result = conn.execute_query(
... "MATCH (p:person {name: 'Denis Villeneuve'})"
... " CREATE (p)-[:directed]->(:movie {title: 'Arrival', year: 2016})",
... counts_=True,
... )
>>> result.counts
GraphWriteCounts(inserted_vertices=1, inserted_edges=1, deleted_vertices=0, deleted_edges=0, updated_properties=0)
There are five counters: inserted_vertices, inserted_edges, deleted_vertices,
deleted_edges and updated_properties. counts.total is their sum, and counts.complete says
whether all five are known.
The server keeps these counters per session, and the driver reads them with a statement of its own,
so asking costs a round trip, or two for a statement that returns rows. Without counts_=True
nothing is read and every counter is None, not 0.
A counter can be None even when you asked. The counters live on the session, and after a write
that returns rows the server has reset only those of the clauses the statement holds; the others
still hold what an earlier statement left there. The driver reads the counters before and after,
and decides each one from the statement’s clauses: a counter no clause of the statement can move is
0, and one that changed was changed by this statement. The case left is a counter one of its
clauses could have moved that did not change and is not zero: a stale number and a real one look
the same there, so it is None rather than a guess. Here SET cannot insert or delete, so those
counters are 0:
>>> conn.execute_query(
... "MATCH (m:movie {title: 'Arrival'}) SET m.genre = 'Sci-Fi' RETURN m.title", counts_=True
... ).counts
GraphWriteCounts(inserted_vertices=0, inserted_edges=0, deleted_vertices=0, deleted_edges=0, updated_properties=1)
A write that returns no rows has all five counters reset by the server, so its counts are always complete.
Many statements at once#
When a batch’s cost is round trips rather than work, a pipeline sends many statements without waiting for each answer.
pipeline_query() is for reads. It returns one Result per statement, in order:
>>> results = conn.pipeline_query([
... ("MATCH (p:person {name: %s}) RETURN p.born", (name,))
... for name in ["Keanu Reeves", "Tom Cruise", "Sam Neill"]
... ])
>>> [r.records for r in results]
[[(1964,)], [(1962,)], [(1947,)]]
It sends 100 statements per pipeline by default (chunk=), measured at about three times the speed
of running them one at a time. If one fails, the driver runs that chunk again one statement at a time,
so the exception you get comes from the statement that caused it. That is also why writes do not
belong in it: a write before the failing statement could run twice. Its results carry no counts.
pipeline_batch() is for writes. It sends every statement, reads nothing back, and returns
None:
>>> conn.pipeline_batch([
... ("CREATE (:movie %s)", ({"title": title, "year": year},))
... for title, year in [("Dune", 2021), ("Dune: Part Two", 2024)]
... ])
>>> conn.execute_query(
... "MATCH (m:movie) WHERE m.title STARTS WITH 'Dune' RETURN count(m)"
... ).records
[(2,)]
Each item is a statement, or a pair of a statement and its parameters. If any statement fails, none
of them is applied, and the failure is raised as agensgraph.errors.BatchFailed. A pipeline does
not say which statement failed: with several statements and only one bad, the server reports the
error against whichever it was working on, which need not be the bad one. So the exception does not
pretend to know. It carries every statement it sent, in order, as statements, and the server’s
error as its __cause__; run them one at a time to find the one at fault. On a connection that is
not in autocommit, roll back afterwards, as after any failure.
For loading many vertices or edges, bulk loading is faster still,
because it uses COPY rather than statements.
One statement per call#
A statement sent with no parameters goes over PostgreSQL’s simple protocol, which runs every statement in the text, separated by semicolons, and reports only the first one’s result. The driver does not split or check what you send. Text you wrote yourself is not a concern; text from somewhere else is, and Untrusted statements shows how to refuse anything but one statement.