1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
|
.. currentmodule:: psycopg
.. _static-typing:
Static Typing
=============
Psycopg source code is annotated according to :pep:`0484` type hints and is
checked using the current version of Mypy_ in ``--strict`` mode.
If your application is checked using Mypy too you can make use of Psycopg
types to validate the correct use of Psycopg objects and of the data returned
by the database.
.. _Mypy: http://mypy-lang.org/
Generic types
-------------
Psycopg `Connection` and `Cursor` objects are `~typing.Generic` objects and
support a `!Row` parameter which is the type of the records returned.
By default methods such as `Cursor.fetchall()` return normal tuples of unknown
size and content. As such, the `connect()` function returns an object of type
`!psycopg.Connection[Tuple[Any, ...]]` and `Connection.cursor()` returns an
object of type `!psycopg.Cursor[Tuple[Any, ...]]`. If you are writing generic
plumbing code it might be practical to use annotations such as
`!Connection[Any]` and `!Cursor[Any]`.
.. code:: python
conn = psycopg.connect() # type is psycopg.Connection[Tuple[Any, ...]]
cur = conn.cursor() # type is psycopg.Cursor[Tuple[Any, ...]]
rec = cur.fetchone() # type is Optional[Tuple[Any, ...]]
recs = cur.fetchall() # type is List[Tuple[Any, ...]]
.. _row-factory-static:
Type of rows returned
---------------------
If you want to use connections and cursors returning your data as different
types, for instance as dictionaries, you can use the `!row_factory` argument
of the `~Connection.connect()` and the `~Connection.cursor()` method, which
will control what type of record is returned by the fetch methods of the
cursors and annotate the returned objects accordingly. See
:ref:`row-factories` for more details.
.. code:: python
dconn = psycopg.connect(row_factory=dict_row)
# dconn type is psycopg.Connection[Dict[str, Any]]
dcur = conn.cursor(row_factory=dict_row)
dcur = dconn.cursor()
# dcur type is psycopg.Cursor[Dict[str, Any]] in both cases
drec = dcur.fetchone()
# drec type is Optional[Dict[str, Any]]
.. _example-pydantic:
Example: returning records as Pydantic models
---------------------------------------------
Using Pydantic_ it is possible to enforce static typing at runtime. Using a
Pydantic model factory the code can be checked statically using Mypy and
querying the database will raise an exception if the rows returned is not
compatible with the model.
.. _Pydantic: https://pydantic-docs.helpmanual.io/
The following example can be checked with ``mypy --strict`` without reporting
any issue. Pydantic will also raise a runtime error in case the
`!Person` is used with a query that returns incompatible data.
.. code:: python
from datetime import date
from typing import Optional
import psycopg
from psycopg.rows import class_row
from pydantic import BaseModel
class Person(BaseModel):
id: int
first_name: str
last_name: str
dob: Optional[date]
def fetch_person(id: int) -> Person:
with psycopg.connect() as conn:
with conn.cursor(row_factory=class_row(Person)) as cur:
cur.execute(
"""
SELECT id, first_name, last_name, dob
FROM (VALUES
(1, 'John', 'Doe', '2000-01-01'::date),
(2, 'Jane', 'White', NULL)
) AS data (id, first_name, last_name, dob)
WHERE id = %(id)s;
""",
{"id": id},
)
obj = cur.fetchone()
# reveal_type(obj) would return 'Optional[Person]' here
if not obj:
raise KeyError(f"person {id} not found")
# reveal_type(obj) would return 'Person' here
return obj
for id in [1, 2]:
p = fetch_person(id)
if p.dob:
print(f"{p.first_name} was born in {p.dob.year}")
else:
print(f"Who knows when {p.first_name} was born")
.. _literal-string:
Checking literal strings in queries
-----------------------------------
The `~Cursor.execute()` method and similar should only receive a literal
string as input, according to :pep:`675`. This means that the query should
come from a literal string in your code, not from an arbitrary string
expression.
For instance, passing an argument to the query should be done via the second
argument to `!execute()`, not by string composition:
.. code:: python
def get_record(conn: psycopg.Connection[Any], id: int) -> Any:
cur = conn.execute("SELECT * FROM my_table WHERE id = %s" % id) # BAD!
return cur.fetchone()
# the function should be implemented as:
def get_record(conn: psycopg.Connection[Any], id: int) -> Any:
cur = conn.execute("select * FROM my_table WHERE id = %s", (id,))
return cur.fetchone()
If you are composing a query dynamically you should use the `sql.SQL` object
and similar to escape safely table and field names. The parameter of the
`!SQL()` object should be a literal string:
.. code:: python
def count_records(conn: psycopg.Connection[Any], table: str) -> int:
query = "SELECT count(*) FROM %s" % table # BAD!
return conn.execute(query).fetchone()[0]
# the function should be implemented as:
def count_records(conn: psycopg.Connection[Any], table: str) -> int:
query = sql.SQL("SELECT count(*) FROM {}").format(sql.Identifier(table))
return conn.execute(query).fetchone()[0]
At the time of writing, no Python static analyzer implements this check (`mypy
doesn't implement it`__, Pyre_ does, but `doesn't work with psycopg yet`__).
Once the type checkers support will be complete, the above bad statements
should be reported as errors.
.. __: https://github.com/python/mypy/issues/12554
.. __: https://github.com/facebook/pyre-check/issues/636
.. _Pyre: https://pyre-check.org/
|
