Leapcell: The Next-Gen Serverless Platform for Python app Hosting
SQLAlchemy Tutorial
SQLAlchemy is the most popular Object Relational Mapping (ORM) in the Python ecosystem. It has an elegant design and is divided into two parts: the underlying Core and the upper-level traditional ORM. In most ORMs in Python and even in other languages, a good hierarchical design has not been implemented. For example, in Django’s ORM, the database connection and the ORM itself are completely mixed together.
Why Do We Need the Core?
The Core layer mainly implements the client connection pool. As the core of modern web applications, the concurrent connection capability of relational databases is often not strong. It is generally not recommended to use a large number of short connections, and in most cases, a connection pool is needed. There are roughly two types of connection pools:
- Server-side connection pool: A specialized connection pool middleware that allocates a long connection for reuse each time for a short connection.
- Client-side connection pool: Generally introduced into the code as a third-party library.
The connection pool of SQLAlchemy belongs to the client-side connection pool. In this connection pool, SQLAlchemy maintains a certain number of long connections. When connect is called, it actually retrieves a connection from the pool; when close is called, it actually returns the connection to the pool.
Creating a Connection
In SQLAlchemy, use create_engine to create a connection (pool). The parameter of create_engine is the URL of the database.
<span>from</span> <span>sqlalchemy</span> <span>import</span> <span>create_engine</span><span># MySQL connection example </span><span>engine</span> <span>=</span> <span>create_engine</span><span>(</span><span>"</span><span>mysql://user:password@localhost:3306/dbname</span><span>"</span><span>,</span><span>echo</span><span>=</span><span>True</span><span>,</span> <span># Setting echo to True will print the actual executed SQL, which is more convenient for debugging </span> <span>future</span><span>=</span><span>True</span><span>,</span> <span># Use the SQLAlchemy 2.0 API, which is backward-compatible </span> <span>pool_size</span><span>=</span><span>5</span><span>,</span> <span># The size of the connection pool is 5 by default. Setting it to 0 means there is no limit to the connection </span> <span>pool_recycle</span><span>=</span><span>3600</span> <span># Set the time to limit the automatic disconnection of the database </span><span>)</span><span># Create an in-memory SQLite database. You must add check_same_thread=False, otherwise it cannot be used in a multithreaded environment </span><span>engine</span> <span>=</span> <span>create_engine</span><span>(</span><span>"</span><span>sqlite:///:memory:</span><span>"</span><span>,</span> <span>echo</span><span>=</span><span>True</span><span>,</span> <span>future</span><span>=</span><span>True</span><span>,</span><span>connect_args</span><span>=</span><span>{</span><span>"</span><span>check_same_thread</span><span>"</span><span>:</span> <span>False</span><span>})</span><span># Another way to connect to MySQL # pip install mysqlclient </span><span>engine</span> <span>=</span> <span>create_engine</span><span>(</span><span>'</span><span>mysql+mysqldb://user:password@localhost/foo?charset=utf8mb4</span><span>'</span><span>)</span><span>from</span> <span>sqlalchemy</span> <span>import</span> <span>create_engine</span> <span># MySQL connection example </span><span>engine</span> <span>=</span> <span>create_engine</span><span>(</span> <span>"</span><span>mysql://user:password@localhost:3306/dbname</span><span>"</span><span>,</span> <span>echo</span><span>=</span><span>True</span><span>,</span> <span># Setting echo to True will print the actual executed SQL, which is more convenient for debugging </span> <span>future</span><span>=</span><span>True</span><span>,</span> <span># Use the SQLAlchemy 2.0 API, which is backward-compatible </span> <span>pool_size</span><span>=</span><span>5</span><span>,</span> <span># The size of the connection pool is 5 by default. Setting it to 0 means there is no limit to the connection </span> <span>pool_recycle</span><span>=</span><span>3600</span> <span># Set the time to limit the automatic disconnection of the database </span><span>)</span> <span># Create an in-memory SQLite database. You must add check_same_thread=False, otherwise it cannot be used in a multithreaded environment </span><span>engine</span> <span>=</span> <span>create_engine</span><span>(</span><span>"</span><span>sqlite:///:memory:</span><span>"</span><span>,</span> <span>echo</span><span>=</span><span>True</span><span>,</span> <span>future</span><span>=</span><span>True</span><span>,</span> <span>connect_args</span><span>=</span><span>{</span><span>"</span><span>check_same_thread</span><span>"</span><span>:</span> <span>False</span><span>})</span> <span># Another way to connect to MySQL # pip install mysqlclient </span><span>engine</span> <span>=</span> <span>create_engine</span><span>(</span><span>'</span><span>mysql+mysqldb://user:password@localhost/foo?charset=utf8mb4</span><span>'</span><span>)</span>from sqlalchemy import create_engine # MySQL connection example engine = create_engine( "mysql://user:password@localhost:3306/dbname", echo=True, # Setting echo to True will print the actual executed SQL, which is more convenient for debugging future=True, # Use the SQLAlchemy 2.0 API, which is backward-compatible pool_size=5, # The size of the connection pool is 5 by default. Setting it to 0 means there is no limit to the connection pool_recycle=3600 # Set the time to limit the automatic disconnection of the database ) # Create an in-memory SQLite database. You must add check_same_thread=False, otherwise it cannot be used in a multithreaded environment engine = create_engine("sqlite:///:memory:", echo=True, future=True, connect_args={"check_same_thread": False}) # Another way to connect to MySQL # pip install mysqlclient engine = create_engine('mysql+mysqldb://user:password@localhost/foo?charset=utf8mb4')
Enter fullscreen mode Exit fullscreen mode
The Core Layer — Using SQL Directly
CRUD
<span>from</span> <span>sqlalchemy</span> <span>import</span> <span>text</span><span>with</span> <span>engine</span><span>.</span><span>connect</span><span>()</span> <span>as</span> <span>conn</span><span>:</span><span>result</span> <span>=</span> <span>conn</span><span>.</span><span>execute</span><span>(</span><span>text</span><span>(</span><span>"</span><span>select * from users</span><span>"</span><span>))</span><span>print</span><span>(</span><span>result</span><span>.</span><span>all</span><span>())</span><span># The result can be iterated over, and each row result is a Row object </span><span>for</span> <span>row</span> <span>in</span> <span>result</span><span>:</span><span># The row object supports three access methods </span> <span>print</span><span>(</span><span>row</span><span>.</span><span>x</span><span>,</span> <span>row</span><span>.</span><span>y</span><span>)</span><span>print</span><span>(</span><span>row</span><span>[</span><span>0</span><span>],</span> <span>row</span><span>[</span><span>1</span><span>])</span><span>print</span><span>(</span><span>row</span><span>[</span><span>"</span><span>x</span><span>"</span><span>],</span> <span>row</span><span>[</span><span>"</span><span>y</span><span>"</span><span>])</span><span># Pass parameters, use `:var` to pass </span><span>result</span> <span>=</span> <span>conn</span><span>.</span><span>execute</span><span>(</span><span>text</span><span>(</span><span>"</span><span>SELECT x, y FROM some_table WHERE y > :y</span><span>"</span><span>),</span><span>{</span><span>"</span><span>y</span><span>"</span><span>:</span> <span>2</span><span>}</span><span>)</span><span># You can also pre-compile the parameters </span><span>stmt</span> <span>=</span> <span>text</span><span>(</span><span>"</span><span>SELECT x, y FROM some_table WHERE y > :y ORDER BY x, y</span><span>"</span><span>).</span><span>bindparams</span><span>(</span><span>y</span><span>=</span><span>6</span><span>)</span><span># When inserting, you can directly insert multiple rows </span><span>conn</span><span>.</span><span>execute</span><span>(</span><span>text</span><span>(</span><span>"</span><span>INSERT INTO some_table (x, y) VALUES (:x, :y)</span><span>"</span><span>),</span><span>[{</span><span>"</span><span>x</span><span>"</span><span>:</span> <span>11</span><span>,</span> <span>"</span><span>y</span><span>"</span><span>:</span> <span>12</span><span>},</span> <span>{</span><span>"</span><span>x</span><span>"</span><span>:</span> <span>13</span><span>,</span> <span>"</span><span>y</span><span>"</span><span>:</span> <span>14</span><span>}]</span><span>)</span><span>from</span> <span>sqlalchemy</span> <span>import</span> <span>text</span> <span>with</span> <span>engine</span><span>.</span><span>connect</span><span>()</span> <span>as</span> <span>conn</span><span>:</span> <span>result</span> <span>=</span> <span>conn</span><span>.</span><span>execute</span><span>(</span><span>text</span><span>(</span><span>"</span><span>select * from users</span><span>"</span><span>))</span> <span>print</span><span>(</span><span>result</span><span>.</span><span>all</span><span>())</span> <span># The result can be iterated over, and each row result is a Row object </span><span>for</span> <span>row</span> <span>in</span> <span>result</span><span>:</span> <span># The row object supports three access methods </span> <span>print</span><span>(</span><span>row</span><span>.</span><span>x</span><span>,</span> <span>row</span><span>.</span><span>y</span><span>)</span> <span>print</span><span>(</span><span>row</span><span>[</span><span>0</span><span>],</span> <span>row</span><span>[</span><span>1</span><span>])</span> <span>print</span><span>(</span><span>row</span><span>[</span><span>"</span><span>x</span><span>"</span><span>],</span> <span>row</span><span>[</span><span>"</span><span>y</span><span>"</span><span>])</span> <span># Pass parameters, use `:var` to pass </span><span>result</span> <span>=</span> <span>conn</span><span>.</span><span>execute</span><span>(</span> <span>text</span><span>(</span><span>"</span><span>SELECT x, y FROM some_table WHERE y > :y</span><span>"</span><span>),</span> <span>{</span><span>"</span><span>y</span><span>"</span><span>:</span> <span>2</span><span>}</span> <span>)</span> <span># You can also pre-compile the parameters </span><span>stmt</span> <span>=</span> <span>text</span><span>(</span><span>"</span><span>SELECT x, y FROM some_table WHERE y > :y ORDER BY x, y</span><span>"</span><span>).</span><span>bindparams</span><span>(</span><span>y</span><span>=</span><span>6</span><span>)</span> <span># When inserting, you can directly insert multiple rows </span><span>conn</span><span>.</span><span>execute</span><span>(</span> <span>text</span><span>(</span><span>"</span><span>INSERT INTO some_table (x, y) VALUES (:x, :y)</span><span>"</span><span>),</span> <span>[{</span><span>"</span><span>x</span><span>"</span><span>:</span> <span>11</span><span>,</span> <span>"</span><span>y</span><span>"</span><span>:</span> <span>12</span><span>},</span> <span>{</span><span>"</span><span>x</span><span>"</span><span>:</span> <span>13</span><span>,</span> <span>"</span><span>y</span><span>"</span><span>:</span> <span>14</span><span>}]</span> <span>)</span>from sqlalchemy import text with engine.connect() as conn: result = conn.execute(text("select * from users")) print(result.all()) # The result can be iterated over, and each row result is a Row object for row in result: # The row object supports three access methods print(row.x, row.y) print(row[0], row[1]) print(row["x"], row["y"]) # Pass parameters, use `:var` to pass result = conn.execute( text("SELECT x, y FROM some_table WHERE y > :y"), {"y": 2} ) # You can also pre-compile the parameters stmt = text("SELECT x, y FROM some_table WHERE y > :y ORDER BY x, y").bindparams(y=6) # When inserting, you can directly insert multiple rows conn.execute( text("INSERT INTO some_table (x, y) VALUES (:x, :y)"), [{"x": 11, "y": 12}, {"x": 13, "y": 14}] )
Enter fullscreen mode Exit fullscreen mode
Transactions and Commit
SQLAlchemy provides two ways to commit, one is manual commit, and the other is semi-automatic commit. The official documentation recommends using engine.begin(). There is also a completely automatic autocommit method that commits once for each row, which is not recommended.
<span># "commit as you go" requires manual commit </span><span>with</span> <span>engine</span><span>.</span><span>connect</span><span>()</span> <span>as</span> <span>conn</span><span>:</span><span>conn</span><span>.</span><span>execute</span><span>(</span><span>text</span><span>(</span><span>"</span><span>CREATE TABLE some_table (x int, y int)</span><span>"</span><span>))</span><span>conn</span><span>.</span><span>execute</span><span>(</span><span>text</span><span>(</span><span>"</span><span>INSERT INTO some_table (x, y) VALUES (:x, :y)</span><span>"</span><span>),</span><span>[{</span><span>"</span><span>x</span><span>"</span><span>:</span> <span>1</span><span>,</span> <span>"</span><span>y</span><span>"</span><span>:</span> <span>1</span><span>},</span> <span>{</span><span>"</span><span>x</span><span>"</span><span>:</span> <span>2</span><span>,</span> <span>"</span><span>y</span><span>"</span><span>:</span> <span>4</span><span>}]</span><span>)</span><span>conn</span><span>.</span><span>commit</span><span>()</span> <span># Note the commit here </span><span># "begin once" semi-automatic commit </span><span>with</span> <span>engine</span><span>.</span><span>begin</span><span>()</span> <span>as</span> <span>conn</span><span>:</span><span>conn</span><span>.</span><span>execute</span><span>(</span><span>text</span><span>(</span><span>"</span><span>INSERT INTO some_table (x, y) VALUES (:x, :y)</span><span>"</span><span>),</span><span>[{</span><span>"</span><span>x</span><span>"</span><span>:</span> <span>6</span><span>,</span> <span>"</span><span>y</span><span>"</span><span>:</span> <span>8</span><span>},</span> <span>{</span><span>"</span><span>x</span><span>"</span><span>:</span> <span>9</span><span>,</span> <span>"</span><span>y</span><span>"</span><span>:</span> <span>10</span><span>}]</span><span>)</span><span># "commit as you go" requires manual commit </span><span>with</span> <span>engine</span><span>.</span><span>connect</span><span>()</span> <span>as</span> <span>conn</span><span>:</span> <span>conn</span><span>.</span><span>execute</span><span>(</span><span>text</span><span>(</span><span>"</span><span>CREATE TABLE some_table (x int, y int)</span><span>"</span><span>))</span> <span>conn</span><span>.</span><span>execute</span><span>(</span> <span>text</span><span>(</span><span>"</span><span>INSERT INTO some_table (x, y) VALUES (:x, :y)</span><span>"</span><span>),</span> <span>[{</span><span>"</span><span>x</span><span>"</span><span>:</span> <span>1</span><span>,</span> <span>"</span><span>y</span><span>"</span><span>:</span> <span>1</span><span>},</span> <span>{</span><span>"</span><span>x</span><span>"</span><span>:</span> <span>2</span><span>,</span> <span>"</span><span>y</span><span>"</span><span>:</span> <span>4</span><span>}]</span> <span>)</span> <span>conn</span><span>.</span><span>commit</span><span>()</span> <span># Note the commit here </span> <span># "begin once" semi-automatic commit </span><span>with</span> <span>engine</span><span>.</span><span>begin</span><span>()</span> <span>as</span> <span>conn</span><span>:</span> <span>conn</span><span>.</span><span>execute</span><span>(</span> <span>text</span><span>(</span><span>"</span><span>INSERT INTO some_table (x, y) VALUES (:x, :y)</span><span>"</span><span>),</span> <span>[{</span><span>"</span><span>x</span><span>"</span><span>:</span> <span>6</span><span>,</span> <span>"</span><span>y</span><span>"</span><span>:</span> <span>8</span><span>},</span> <span>{</span><span>"</span><span>x</span><span>"</span><span>:</span> <span>9</span><span>,</span> <span>"</span><span>y</span><span>"</span><span>:</span> <span>10</span><span>}]</span> <span>)</span># "commit as you go" requires manual commit with engine.connect() as conn: conn.execute(text("CREATE TABLE some_table (x int, y int)")) conn.execute( text("INSERT INTO some_table (x, y) VALUES (:x, :y)"), [{"x": 1, "y": 1}, {"x": 2, "y": 4}] ) conn.commit() # Note the commit here # "begin once" semi-automatic commit with engine.begin() as conn: conn.execute( text("INSERT INTO some_table (x, y) VALUES (:x, :y)"), [{"x": 6, "y": 8}, {"x": 9, "y": 10}] )
Enter fullscreen mode Exit fullscreen mode
ORM
Session
The Session is not thread-safe. But generally, the web framework should obtain a session at the start of each request, so it is not a problem either.
<span>from</span> <span>sqlalchemy.orm</span> <span>import</span> <span>Session</span><span>with</span> <span>Session</span><span>(</span><span>engine</span><span>)</span> <span>as</span> <span>session</span><span>:</span><span>session</span><span>.</span><span>add</span><span>(</span><span>foo</span><span>)</span><span>session</span><span>.</span><span>commit</span><span>()</span><span># You can also use sessionmaker to create a factory function, so you don't have to enter parameters every time </span><span>from</span> <span>sqlalchemy.orm</span> <span>import</span> <span>sessionmaker</span><span>new_session</span> <span>=</span> <span>sessionmaker</span><span>(</span><span>engine</span><span>)</span><span>with</span> <span>new_session</span><span>()</span> <span>as</span> <span>session</span><span>:</span><span>...</span><span>from</span> <span>sqlalchemy.orm</span> <span>import</span> <span>Session</span> <span>with</span> <span>Session</span><span>(</span><span>engine</span><span>)</span> <span>as</span> <span>session</span><span>:</span> <span>session</span><span>.</span><span>add</span><span>(</span><span>foo</span><span>)</span> <span>session</span><span>.</span><span>commit</span><span>()</span> <span># You can also use sessionmaker to create a factory function, so you don't have to enter parameters every time </span><span>from</span> <span>sqlalchemy.orm</span> <span>import</span> <span>sessionmaker</span> <span>new_session</span> <span>=</span> <span>sessionmaker</span><span>(</span><span>engine</span><span>)</span> <span>with</span> <span>new_session</span><span>()</span> <span>as</span> <span>session</span><span>:</span> <span>...</span>from sqlalchemy.orm import Session with Session(engine) as session: session.add(foo) session.commit() # You can also use sessionmaker to create a factory function, so you don't have to enter parameters every time from sqlalchemy.orm import sessionmaker new_session = sessionmaker(engine) with new_session() as session: ...
Enter fullscreen mode Exit fullscreen mode
Declarative API
- Use
__tablename__to specify the database table name. - Use
Mappedand native types to declare each field. - Use
Integer,String, etc. to specify the field type. - Use the
indexparameter to specify the index. - Use the
uniqueparameter to specify the unique index. - Use
__table_args__to specify other attributes, such as composite indexes.
<span>from</span> <span>datetime</span> <span>import</span> <span>datetime</span><span>from</span> <span>sqlalchemy</span> <span>import</span> <span>Integer</span><span>,</span> <span>String</span><span>,</span> <span>func</span><span>,</span> <span>UniqueConstraint</span><span>from</span> <span>sqlalchemy.orm</span> <span>import</span> <span>relationship</span><span>,</span> <span>mapped_column</span><span>,</span> <span>Mapped</span><span>from</span> <span>sqlalchemy.orm</span> <span>import</span> <span>DeclarativeBase</span><span>class</span> <span>Base</span><span>(</span><span>DeclarativeBase</span><span>):</span><span>pass</span><span>class</span> <span>User</span><span>(</span><span>Base</span><span>):</span><span>__tablename__</span> <span>=</span> <span>"</span><span>users</span><span>"</span><span># It must be a tuple, not a list </span> <span>__table_args__</span> <span>=</span> <span>(</span><span>UniqueConstraint</span><span>(</span><span>"</span><span>name</span><span>"</span><span>,</span> <span>"</span><span>time_created</span><span>"</span><span>),)</span><span>id</span><span>:</span> <span>Mapped</span><span>[</span><span>int</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>Integer</span><span>,</span> <span>primary_key</span><span>=</span><span>True</span><span>)</span><span>name</span><span>:</span> <span>Mapped</span><span>[</span><span>str</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>String</span><span>(</span><span>30</span><span>),</span> <span>index</span><span>=</span><span>True</span><span>)</span><span>fullname</span><span>:</span> <span>Mapped</span><span>[</span><span>str</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>String</span><span>,</span> <span>unique</span><span>=</span><span>True</span><span>)</span><span># For particularly large fields, you can also use deferred, so that this field is not loaded by default </span> <span>description</span><span>:</span> <span>Mapped</span><span>[</span><span>str</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>Text</span><span>,</span> <span>deferred</span><span>=</span><span>True</span><span>)</span><span># Default value, note that a function is passed, not the current time </span> <span>time_created</span><span>:</span> <span>Mapped</span><span>[</span><span>datetime</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>DateTime</span><span>(</span><span>Timezone</span><span>=</span><span>True</span><span>),</span> <span>default</span><span>=</span><span>datetime</span><span>.</span><span>now</span><span>)</span><span># Or use the server default value, but it must be set when the table is created and will become part of the table's schema </span> <span>time_created</span><span>:</span> <span>Mapped</span><span>[</span><span>datetime</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>DateTime</span><span>(</span><span>timezone</span><span>=</span><span>True</span><span>),</span> <span>server_default</span><span>=</span><span>func</span><span>.</span><span>now</span><span>())</span><span>time_updated</span><span>:</span> <span>Mapped</span><span>[</span><span>datetime</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>DateTime</span><span>(</span><span>timezone</span><span>=</span><span>True</span><span>),</span> <span>onupdate</span><span>=</span><span>func</span><span>.</span><span>now</span><span>())</span><span>class</span> <span>Address</span><span>(</span><span>Base</span><span>):</span><span>__tablename__</span> <span>=</span> <span>"</span><span>address</span><span>"</span><span>id</span><span>:</span> <span>Mapped</span><span>[</span><span>int</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>Integer</span><span>,</span> <span>primary_key</span><span>=</span><span>True</span><span>)</span><span>email_address</span><span>:</span> <span>Mapped</span><span>[</span><span>str</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>String</span><span>,</span> <span>nullable</span><span>=</span><span>False</span><span>)</span><span># Call create_all to create all models </span><span>Base</span><span>.</span><span>metadata</span><span>.</span><span>create_all</span><span>(</span><span>engine</span><span>)</span><span># If you only need to create one model </span><span>User</span><span>.</span><span>__table__</span><span>.</span><span>create</span><span>(</span><span>engine</span><span>)</span><span>from</span> <span>datetime</span> <span>import</span> <span>datetime</span> <span>from</span> <span>sqlalchemy</span> <span>import</span> <span>Integer</span><span>,</span> <span>String</span><span>,</span> <span>func</span><span>,</span> <span>UniqueConstraint</span> <span>from</span> <span>sqlalchemy.orm</span> <span>import</span> <span>relationship</span><span>,</span> <span>mapped_column</span><span>,</span> <span>Mapped</span> <span>from</span> <span>sqlalchemy.orm</span> <span>import</span> <span>DeclarativeBase</span> <span>class</span> <span>Base</span><span>(</span><span>DeclarativeBase</span><span>):</span> <span>pass</span> <span>class</span> <span>User</span><span>(</span><span>Base</span><span>):</span> <span>__tablename__</span> <span>=</span> <span>"</span><span>users</span><span>"</span> <span># It must be a tuple, not a list </span> <span>__table_args__</span> <span>=</span> <span>(</span><span>UniqueConstraint</span><span>(</span><span>"</span><span>name</span><span>"</span><span>,</span> <span>"</span><span>time_created</span><span>"</span><span>),)</span> <span>id</span><span>:</span> <span>Mapped</span><span>[</span><span>int</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>Integer</span><span>,</span> <span>primary_key</span><span>=</span><span>True</span><span>)</span> <span>name</span><span>:</span> <span>Mapped</span><span>[</span><span>str</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>String</span><span>(</span><span>30</span><span>),</span> <span>index</span><span>=</span><span>True</span><span>)</span> <span>fullname</span><span>:</span> <span>Mapped</span><span>[</span><span>str</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>String</span><span>,</span> <span>unique</span><span>=</span><span>True</span><span>)</span> <span># For particularly large fields, you can also use deferred, so that this field is not loaded by default </span> <span>description</span><span>:</span> <span>Mapped</span><span>[</span><span>str</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>Text</span><span>,</span> <span>deferred</span><span>=</span><span>True</span><span>)</span> <span># Default value, note that a function is passed, not the current time </span> <span>time_created</span><span>:</span> <span>Mapped</span><span>[</span><span>datetime</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>DateTime</span><span>(</span><span>Timezone</span><span>=</span><span>True</span><span>),</span> <span>default</span><span>=</span><span>datetime</span><span>.</span><span>now</span><span>)</span> <span># Or use the server default value, but it must be set when the table is created and will become part of the table's schema </span> <span>time_created</span><span>:</span> <span>Mapped</span><span>[</span><span>datetime</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>DateTime</span><span>(</span><span>timezone</span><span>=</span><span>True</span><span>),</span> <span>server_default</span><span>=</span><span>func</span><span>.</span><span>now</span><span>())</span> <span>time_updated</span><span>:</span> <span>Mapped</span><span>[</span><span>datetime</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>DateTime</span><span>(</span><span>timezone</span><span>=</span><span>True</span><span>),</span> <span>onupdate</span><span>=</span><span>func</span><span>.</span><span>now</span><span>())</span> <span>class</span> <span>Address</span><span>(</span><span>Base</span><span>):</span> <span>__tablename__</span> <span>=</span> <span>"</span><span>address</span><span>"</span> <span>id</span><span>:</span> <span>Mapped</span><span>[</span><span>int</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>Integer</span><span>,</span> <span>primary_key</span><span>=</span><span>True</span><span>)</span> <span>email_address</span><span>:</span> <span>Mapped</span><span>[</span><span>str</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>String</span><span>,</span> <span>nullable</span><span>=</span><span>False</span><span>)</span> <span># Call create_all to create all models </span><span>Base</span><span>.</span><span>metadata</span><span>.</span><span>create_all</span><span>(</span><span>engine</span><span>)</span> <span># If you only need to create one model </span><span>User</span><span>.</span><span>__table__</span><span>.</span><span>create</span><span>(</span><span>engine</span><span>)</span>from datetime import datetime from sqlalchemy import Integer, String, func, UniqueConstraint from sqlalchemy.orm import relationship, mapped_column, Mapped from sqlalchemy.orm import DeclarativeBase class Base(DeclarativeBase): pass class User(Base): __tablename__ = "users" # It must be a tuple, not a list __table_args__ = (UniqueConstraint("name", "time_created"),) id: Mapped[int] = mapped_column(Integer, primary_key=True) name: Mapped[str] = mapped_column(String(30), index=True) fullname: Mapped[str] = mapped_column(String, unique=True) # For particularly large fields, you can also use deferred, so that this field is not loaded by default description: Mapped[str] = mapped_column(Text, deferred=True) # Default value, note that a function is passed, not the current time time_created: Mapped[datetime] = mapped_column(DateTime(Timezone=True), default=datetime.now) # Or use the server default value, but it must be set when the table is created and will become part of the table's schema time_created: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) time_updated: Mapped[datetime] = mapped_column(DateTime(timezone=True), onupdate=func.now()) class Address(Base): __tablename__ = "address" id: Mapped[int] = mapped_column(Integer, primary_key=True) email_address: Mapped[str] = mapped_column(String, nullable=False) # Call create_all to create all models Base.metadata.create_all(engine) # If you only need to create one model User.__table__.create(engine)
Enter fullscreen mode Exit fullscreen mode
Foreign Keys
Use relationship to specify the association relationship between models.
Bi-directional Mapping of One-to-Many Relationship
<span>from</span> <span>sqlalchemy</span> <span>import</span> <span>create_engine</span><span>,</span> <span>Integer</span><span>,</span> <span>String</span><span>,</span> <span>ForeignKey</span><span>from</span> <span>sqlalchemy.orm</span> <span>import</span> <span>DeclarativeBase</span><span>,</span> <span>relationship</span><span>,</span> <span>Session</span><span>,</span> <span>Mapped</span><span>,</span> <span>mapped_column</span><span>class</span> <span>Group</span><span>(</span><span>Base</span><span>):</span><span>__tablename__</span> <span>=</span> <span>'</span><span>groups</span><span>'</span><span>id</span><span>:</span> <span>Mapped</span><span>[</span><span>int</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>Integer</span><span>,</span> <span>primary_key</span><span>=</span><span>True</span><span>)</span><span>name</span><span>:</span> <span>Mapped</span><span>[</span><span>str</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>String</span><span>)</span><span># The corresponding multiple users, here use the model name as the parameter </span> <span>members</span> <span>=</span> <span>relationship</span><span>(</span><span>'</span><span>User</span><span>'</span><span>)</span><span>class</span> <span>User</span><span>(</span><span>Base</span><span>):</span><span>__tablename__</span> <span>=</span> <span>'</span><span>users</span><span>'</span><span>id</span> <span>=</span> <span>Column</span><span>(</span><span>Integer</span><span>,</span> <span>primary_key</span><span>=</span><span>True</span><span>)</span><span>name</span> <span>=</span> <span>Column</span><span>(</span><span>String</span><span>)</span><span># group_id is the real foreign key name in the database, and the second field ForeignKey is used to specify the corresponding ID </span> <span>group_id</span> <span>=</span> <span>Column</span><span>(</span><span>Integer</span><span>,</span> <span>ForeignKey</span><span>(</span><span>'</span><span>groups.id</span><span>'</span><span>))</span><span># The corresponding group field in the model, which needs to declare which field in the corresponding model it overlaps with </span> <span>group</span> <span>=</span> <span>relationship</span><span>(</span><span>'</span><span>Group</span><span>'</span><span>,</span> <span>overlaps</span><span>=</span><span>"</span><span>members</span><span>"</span><span>)</span><span>from</span> <span>sqlalchemy</span> <span>import</span> <span>create_engine</span><span>,</span> <span>Integer</span><span>,</span> <span>String</span><span>,</span> <span>ForeignKey</span> <span>from</span> <span>sqlalchemy.orm</span> <span>import</span> <span>DeclarativeBase</span><span>,</span> <span>relationship</span><span>,</span> <span>Session</span><span>,</span> <span>Mapped</span><span>,</span> <span>mapped_column</span> <span>class</span> <span>Group</span><span>(</span><span>Base</span><span>):</span> <span>__tablename__</span> <span>=</span> <span>'</span><span>groups</span><span>'</span> <span>id</span><span>:</span> <span>Mapped</span><span>[</span><span>int</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>Integer</span><span>,</span> <span>primary_key</span><span>=</span><span>True</span><span>)</span> <span>name</span><span>:</span> <span>Mapped</span><span>[</span><span>str</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>String</span><span>)</span> <span># The corresponding multiple users, here use the model name as the parameter </span> <span>members</span> <span>=</span> <span>relationship</span><span>(</span><span>'</span><span>User</span><span>'</span><span>)</span> <span>class</span> <span>User</span><span>(</span><span>Base</span><span>):</span> <span>__tablename__</span> <span>=</span> <span>'</span><span>users</span><span>'</span> <span>id</span> <span>=</span> <span>Column</span><span>(</span><span>Integer</span><span>,</span> <span>primary_key</span><span>=</span><span>True</span><span>)</span> <span>name</span> <span>=</span> <span>Column</span><span>(</span><span>String</span><span>)</span> <span># group_id is the real foreign key name in the database, and the second field ForeignKey is used to specify the corresponding ID </span> <span>group_id</span> <span>=</span> <span>Column</span><span>(</span><span>Integer</span><span>,</span> <span>ForeignKey</span><span>(</span><span>'</span><span>groups.id</span><span>'</span><span>))</span> <span># The corresponding group field in the model, which needs to declare which field in the corresponding model it overlaps with </span> <span>group</span> <span>=</span> <span>relationship</span><span>(</span><span>'</span><span>Group</span><span>'</span><span>,</span> <span>overlaps</span><span>=</span><span>"</span><span>members</span><span>"</span><span>)</span>from sqlalchemy import create_engine, Integer, String, ForeignKey from sqlalchemy.orm import DeclarativeBase, relationship, Session, Mapped, mapped_column class Group(Base): __tablename__ = 'groups' id: Mapped[int] = mapped_column(Integer, primary_key=True) name: Mapped[str] = mapped_column(String) # The corresponding multiple users, here use the model name as the parameter members = relationship('User') class User(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True) name = Column(String) # group_id is the real foreign key name in the database, and the second field ForeignKey is used to specify the corresponding ID group_id = Column(Integer, ForeignKey('groups.id')) # The corresponding group field in the model, which needs to declare which field in the corresponding model it overlaps with group = relationship('Group', overlaps="members")
Enter fullscreen mode Exit fullscreen mode
Many-to-Many Mapping, an Association Table is Required
<span># Association table </span><span>class</span> <span>UserPermissions</span><span>(</span><span>Base</span><span>):</span><span>__tablename__</span> <span>=</span> <span>'</span><span>user_permissions</span><span>'</span><span>id</span><span>:</span> <span>Mapped</span><span>[</span><span>int</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>Integer</span><span>,</span> <span>primary_key</span><span>=</span><span>True</span><span>)</span><span># Also use foreign key to specify the foreign key </span> <span>user_id</span><span>:</span> <span>Mapped</span><span>[</span><span>int</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>Integer</span><span>,</span> <span>ForeignKey</span><span>(</span><span>'</span><span>users.id</span><span>'</span><span>))</span><span>permission_id</span><span>:</span> <span>Mapped</span><span>[</span><span>str</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>String</span><span>,</span> <span>ForeignKey</span><span>(</span><span>'</span><span>permissions.id</span><span>'</span><span>))</span><span>class</span> <span>User</span><span>(</span><span>Base</span><span>):</span><span>__tablename__</span> <span>=</span> <span>'</span><span>users</span><span>'</span><span>id</span><span>:</span> <span>Mapped</span><span>[</span><span>int</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>Integer</span><span>,</span> <span>primary_key</span><span>=</span><span>True</span><span>)</span><span>name</span><span>:</span> <span>Mapped</span><span>[</span><span>str</span><span>]</span> <span>=</span> <span>Column</span><span>(</span><span>String</span><span>)</span><span># Use secondary to specify the association table, and also use overlaps to specify the corresponding field in the model </span> <span>permissions</span> <span>=</span> <span>relationship</span><span>(</span><span>'</span><span>Permission</span><span>'</span><span>,</span> <span>secondary</span><span>=</span><span>"</span><span>user_permissions</span><span>"</span><span>,</span> <span>overlaps</span><span>=</span><span>"</span><span>users</span><span>"</span><span>)</span><span>class</span> <span>Permission</span><span>(</span><span>Base</span><span>):</span><span>__tablename__</span> <span>=</span> <span>'</span><span>permissions</span><span>'</span><span>id</span><span>:</span> <span>Mapped</span><span>[</span><span>int</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>Integer</span><span>,</span> <span>primary_key</span><span>=</span><span>True</span><span>)</span><span>name</span><span>:</span> <span>Mapped</span><span>[</span><span>str</span><span>]</span> <span>=</span> <span>Column</span><span>(</span><span>String</span><span>)</span><span># The same as above </span> <span>users</span> <span>=</span> <span>relationship</span><span>(</span><span>'</span><span>User</span><span>'</span><span>,</span> <span>secondary</span><span>=</span><span>"</span><span>user_permissions</span><span>"</span><span>,</span> <span>overlaps</span><span>=</span><span>"</span><span>permissions</span><span>"</span><span>)</span><span>user1</span> <span>=</span> <span>User</span><span>(</span><span>name</span><span>=</span><span>'</span><span>user1</span><span>'</span><span>,</span> <span>group_id</span><span>=</span><span>1</span><span>)</span><span>user2</span> <span>=</span> <span>User</span><span>(</span><span>name</span><span>=</span><span>'</span><span>user2</span><span>'</span><span>)</span><span>group1</span> <span>=</span> <span>Group</span><span>(</span><span>name</span><span>=</span><span>'</span><span>group1</span><span>'</span><span>)</span><span>group2</span> <span>=</span> <span>Group</span><span>(</span><span>name</span><span>=</span><span>'</span><span>group2</span><span>'</span><span>,</span> <span>members</span><span>=</span><span>[</span><span>user2</span><span>])</span><span>permission1</span> <span>=</span> <span>Permission</span><span>(</span><span>name</span><span>=</span><span>"</span><span>open_file</span><span>"</span><span>)</span><span>permission2</span> <span>=</span> <span>Permission</span><span>(</span><span>name</span><span>=</span><span>"</span><span>save_file</span><span>"</span><span>)</span><span>user1</span><span>.</span><span>permissions</span><span>.</span><span>append</span><span>(</span><span>permission1</span><span>)</span><span>db</span><span>.</span><span>add_all</span><span>([</span><span>user1</span><span>,</span> <span>user2</span><span>,</span> <span>group1</span><span>,</span> <span>group2</span><span>,</span> <span>permission1</span><span>,</span> <span>permission2</span><span>])</span><span>db</span><span>.</span><span>commit</span><span>()</span><span>print</span><span>(</span><span>user1</span><span>.</span><span>permissions</span><span>[</span><span>0</span><span>].</span><span>id</span><span>)</span><span># Association table </span><span>class</span> <span>UserPermissions</span><span>(</span><span>Base</span><span>):</span> <span>__tablename__</span> <span>=</span> <span>'</span><span>user_permissions</span><span>'</span> <span>id</span><span>:</span> <span>Mapped</span><span>[</span><span>int</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>Integer</span><span>,</span> <span>primary_key</span><span>=</span><span>True</span><span>)</span> <span># Also use foreign key to specify the foreign key </span> <span>user_id</span><span>:</span> <span>Mapped</span><span>[</span><span>int</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>Integer</span><span>,</span> <span>ForeignKey</span><span>(</span><span>'</span><span>users.id</span><span>'</span><span>))</span> <span>permission_id</span><span>:</span> <span>Mapped</span><span>[</span><span>str</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>String</span><span>,</span> <span>ForeignKey</span><span>(</span><span>'</span><span>permissions.id</span><span>'</span><span>))</span> <span>class</span> <span>User</span><span>(</span><span>Base</span><span>):</span> <span>__tablename__</span> <span>=</span> <span>'</span><span>users</span><span>'</span> <span>id</span><span>:</span> <span>Mapped</span><span>[</span><span>int</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>Integer</span><span>,</span> <span>primary_key</span><span>=</span><span>True</span><span>)</span> <span>name</span><span>:</span> <span>Mapped</span><span>[</span><span>str</span><span>]</span> <span>=</span> <span>Column</span><span>(</span><span>String</span><span>)</span> <span># Use secondary to specify the association table, and also use overlaps to specify the corresponding field in the model </span> <span>permissions</span> <span>=</span> <span>relationship</span><span>(</span><span>'</span><span>Permission</span><span>'</span><span>,</span> <span>secondary</span><span>=</span><span>"</span><span>user_permissions</span><span>"</span><span>,</span> <span>overlaps</span><span>=</span><span>"</span><span>users</span><span>"</span><span>)</span> <span>class</span> <span>Permission</span><span>(</span><span>Base</span><span>):</span> <span>__tablename__</span> <span>=</span> <span>'</span><span>permissions</span><span>'</span> <span>id</span><span>:</span> <span>Mapped</span><span>[</span><span>int</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>Integer</span><span>,</span> <span>primary_key</span><span>=</span><span>True</span><span>)</span> <span>name</span><span>:</span> <span>Mapped</span><span>[</span><span>str</span><span>]</span> <span>=</span> <span>Column</span><span>(</span><span>String</span><span>)</span> <span># The same as above </span> <span>users</span> <span>=</span> <span>relationship</span><span>(</span><span>'</span><span>User</span><span>'</span><span>,</span> <span>secondary</span><span>=</span><span>"</span><span>user_permissions</span><span>"</span><span>,</span> <span>overlaps</span><span>=</span><span>"</span><span>permissions</span><span>"</span><span>)</span> <span>user1</span> <span>=</span> <span>User</span><span>(</span><span>name</span><span>=</span><span>'</span><span>user1</span><span>'</span><span>,</span> <span>group_id</span><span>=</span><span>1</span><span>)</span> <span>user2</span> <span>=</span> <span>User</span><span>(</span><span>name</span><span>=</span><span>'</span><span>user2</span><span>'</span><span>)</span> <span>group1</span> <span>=</span> <span>Group</span><span>(</span><span>name</span><span>=</span><span>'</span><span>group1</span><span>'</span><span>)</span> <span>group2</span> <span>=</span> <span>Group</span><span>(</span><span>name</span><span>=</span><span>'</span><span>group2</span><span>'</span><span>,</span> <span>members</span><span>=</span><span>[</span><span>user2</span><span>])</span> <span>permission1</span> <span>=</span> <span>Permission</span><span>(</span><span>name</span><span>=</span><span>"</span><span>open_file</span><span>"</span><span>)</span> <span>permission2</span> <span>=</span> <span>Permission</span><span>(</span><span>name</span><span>=</span><span>"</span><span>save_file</span><span>"</span><span>)</span> <span>user1</span><span>.</span><span>permissions</span><span>.</span><span>append</span><span>(</span><span>permission1</span><span>)</span> <span>db</span><span>.</span><span>add_all</span><span>([</span><span>user1</span><span>,</span> <span>user2</span><span>,</span> <span>group1</span><span>,</span> <span>group2</span><span>,</span> <span>permission1</span><span>,</span> <span>permission2</span><span>])</span> <span>db</span><span>.</span><span>commit</span><span>()</span> <span>print</span><span>(</span><span>user1</span><span>.</span><span>permissions</span><span>[</span><span>0</span><span>].</span><span>id</span><span>)</span># Association table class UserPermissions(Base): __tablename__ = 'user_permissions' id: Mapped[int] = mapped_column(Integer, primary_key=True) # Also use foreign key to specify the foreign key user_id: Mapped[int] = mapped_column(Integer, ForeignKey('users.id')) permission_id: Mapped[str] = mapped_column(String, ForeignKey('permissions.id')) class User(Base): __tablename__ = 'users' id: Mapped[int] = mapped_column(Integer, primary_key=True) name: Mapped[str] = Column(String) # Use secondary to specify the association table, and also use overlaps to specify the corresponding field in the model permissions = relationship('Permission', secondary="user_permissions", overlaps="users") class Permission(Base): __tablename__ = 'permissions' id: Mapped[int] = mapped_column(Integer, primary_key=True) name: Mapped[str] = Column(String) # The same as above users = relationship('User', secondary="user_permissions", overlaps="permissions") user1 = User(name='user1', group_id=1) user2 = User(name='user2') group1 = Group(name='group1') group2 = Group(name='group2', members=[user2]) permission1 = Permission(name="open_file") permission2 = Permission(name="save_file") user1.permissions.append(permission1) db.add_all([user1, user2, group1, group2, permission1, permission2]) db.commit() print(user1.permissions[0].id)
Enter fullscreen mode Exit fullscreen mode
In most other tutorials, backref is used to generate the attributes of the corresponding model. Here, it is more preferable to explicitly declare the accessible attributes in the corresponding model.
CRUD
Different from the 1.x API, in the 2.0 API, query is no longer used, but select is used to query data.
<span>from</span> <span>sqlalchemy</span> <span>import</span> <span>select</span><span># The parameter of where is an expression composed of `==`. The advantage is that when writing code, spelling errors will be detected </span><span>stmt</span> <span>=</span> <span>select</span><span>(</span><span>User</span><span>).</span><span>where</span><span>(</span><span>User</span><span>.</span><span>name</span> <span>==</span> <span>"</span><span>john</span><span>"</span><span>).</span><span>order_by</span><span>(</span><span>User</span><span>.</span><span>id</span><span>)</span><span># filter_by uses **kwargs as parameters </span><span>stmt</span> <span>=</span> <span>select</span><span>(</span><span>User</span><span>).</span><span>filter_by</span><span>(</span><span>name</span><span>=</span><span>"</span><span>some_user</span><span>"</span><span>)</span><span># order_by can also use User.id.desc() to represent reverse sorting </span><span>result</span> <span>=</span> <span>session</span><span>.</span><span>execute</span><span>(</span><span>stmt</span><span>)</span><span># Generally, when selecting the entire object, the scalars method should be used, otherwise a tuple containing one object will be returned </span><span>for</span> <span>user</span> <span>in</span> <span>result</span><span>.</span><span>scalars</span><span>():</span><span>print</span><span>(</span><span>user</span><span>.</span><span>name</span><span>)</span><span># When querying a single attribute of the model, there is no need to use scalars </span><span>result</span> <span>=</span> <span>session</span><span>.</span><span>execute</span><span>(</span><span>select</span><span>(</span><span>User</span><span>.</span><span>name</span><span>))</span><span>for</span> <span>row</span> <span>in</span> <span>result</span><span>:</span><span>print</span><span>(</span><span>row</span><span>.</span><span>name</span><span>)</span><span># There is also a shortcut to query by id: </span><span>user</span> <span>=</span> <span>session</span><span>.</span><span>get</span><span>(</span><span>User</span><span>,</span> <span>pk</span><span>=</span><span>1</span><span>)</span><span># To update data, the update statement needs to be used </span><span>from</span> <span>sqlalchemy</span> <span>import</span> <span>update</span><span># synchronize_session has three options: false, "fetch", "evaluate", and the default is evaluate # false means not updating the object in Python at all # fetch means reloading an object from the database # evaluate means that while updating the database, the same operation is also tried on the object in Python as much as possible </span><span>stmt</span> <span>=</span> <span>update</span><span>(</span><span>User</span><span>).</span><span>where</span><span>(</span><span>User</span><span>.</span><span>name</span> <span>==</span> <span>"</span><span>john</span><span>"</span><span>).</span><span>values</span><span>(</span><span>name</span><span>=</span><span>"</span><span>John</span><span>"</span><span>).</span><span>execution_options</span><span>(</span><span>synchronize_session</span><span>=</span><span>"</span><span>fetch</span><span>"</span><span>)</span><span>session</span><span>.</span><span>execute</span><span>(</span><span>stmt</span><span>)</span><span># Or directly assign a value to the attribute </span><span>user</span><span>.</span><span>name</span> <span>=</span> <span>"</span><span>John</span><span>"</span><span>session</span><span>.</span><span>commit</span><span>()</span><span># There is a place here that may introduce a race condition (竞态条件) # Wrong! If two processes update this value at the same time, it may result in only one value being updated. # Both assign the value they think is correct, which is 2, but the actual correct value is 1 + 1 + 1 = 3 # Corresponding SQL: Update users set visit_count = 2 where user.id = 1 </span><span>user</span><span>.</span><span>visit_count</span> <span>+=</span> <span>1</span><span># Correct approach: Note the capital U, that is, using the attribute of the model, and the generated SQL is to add 1 on the SQL server side # Corresponding SQL: Update users set visit_count = visit_count + 1 where user.id = 1 </span><span>user</span><span>.</span><span>visit_count</span> <span>=</span> <span>User</span><span>.</span><span>visit_count</span> <span>+</span> <span>1</span><span># To add an object, directly use the session.add method </span><span>session</span><span>.</span><span>add</span><span>(</span><span>user</span><span>)</span><span># Or add_all </span><span>session</span><span>.</span><span>add_all</span><span>([</span><span>user1</span><span>,</span> <span>user2</span><span>,</span> <span>group1</span><span>])</span><span># If you want to get the inserted ID, of course, you can also read it after committing </span><span>session</span><span>.</span><span>flush</span><span>()</span> <span># flush is not a commit, and the transaction has not been committed. It should be repeatable read, which is related to the isolation level of the database. </span><span>print</span><span>(</span><span>user</span><span>.</span><span>id</span><span>)</span><span># To delete, use session.delete </span><span>session</span><span>.</span><span>delete</span><span>(</span><span>user</span><span>)</span><span>from</span> <span>sqlalchemy</span> <span>import</span> <span>select</span> <span># The parameter of where is an expression composed of `==`. The advantage is that when writing code, spelling errors will be detected </span><span>stmt</span> <span>=</span> <span>select</span><span>(</span><span>User</span><span>).</span><span>where</span><span>(</span><span>User</span><span>.</span><span>name</span> <span>==</span> <span>"</span><span>john</span><span>"</span><span>).</span><span>order_by</span><span>(</span><span>User</span><span>.</span><span>id</span><span>)</span> <span># filter_by uses **kwargs as parameters </span><span>stmt</span> <span>=</span> <span>select</span><span>(</span><span>User</span><span>).</span><span>filter_by</span><span>(</span><span>name</span><span>=</span><span>"</span><span>some_user</span><span>"</span><span>)</span> <span># order_by can also use User.id.desc() to represent reverse sorting </span> <span>result</span> <span>=</span> <span>session</span><span>.</span><span>execute</span><span>(</span><span>stmt</span><span>)</span> <span># Generally, when selecting the entire object, the scalars method should be used, otherwise a tuple containing one object will be returned </span><span>for</span> <span>user</span> <span>in</span> <span>result</span><span>.</span><span>scalars</span><span>():</span> <span>print</span><span>(</span><span>user</span><span>.</span><span>name</span><span>)</span> <span># When querying a single attribute of the model, there is no need to use scalars </span><span>result</span> <span>=</span> <span>session</span><span>.</span><span>execute</span><span>(</span><span>select</span><span>(</span><span>User</span><span>.</span><span>name</span><span>))</span> <span>for</span> <span>row</span> <span>in</span> <span>result</span><span>:</span> <span>print</span><span>(</span><span>row</span><span>.</span><span>name</span><span>)</span> <span># There is also a shortcut to query by id: </span><span>user</span> <span>=</span> <span>session</span><span>.</span><span>get</span><span>(</span><span>User</span><span>,</span> <span>pk</span><span>=</span><span>1</span><span>)</span> <span># To update data, the update statement needs to be used </span><span>from</span> <span>sqlalchemy</span> <span>import</span> <span>update</span> <span># synchronize_session has three options: false, "fetch", "evaluate", and the default is evaluate # false means not updating the object in Python at all # fetch means reloading an object from the database # evaluate means that while updating the database, the same operation is also tried on the object in Python as much as possible </span><span>stmt</span> <span>=</span> <span>update</span><span>(</span><span>User</span><span>).</span><span>where</span><span>(</span><span>User</span><span>.</span><span>name</span> <span>==</span> <span>"</span><span>john</span><span>"</span><span>).</span><span>values</span><span>(</span><span>name</span><span>=</span><span>"</span><span>John</span><span>"</span><span>).</span><span>execution_options</span><span>(</span><span>synchronize_session</span><span>=</span><span>"</span><span>fetch</span><span>"</span><span>)</span> <span>session</span><span>.</span><span>execute</span><span>(</span><span>stmt</span><span>)</span> <span># Or directly assign a value to the attribute </span><span>user</span><span>.</span><span>name</span> <span>=</span> <span>"</span><span>John</span><span>"</span> <span>session</span><span>.</span><span>commit</span><span>()</span> <span># There is a place here that may introduce a race condition (竞态条件) # Wrong! If two processes update this value at the same time, it may result in only one value being updated. # Both assign the value they think is correct, which is 2, but the actual correct value is 1 + 1 + 1 = 3 # Corresponding SQL: Update users set visit_count = 2 where user.id = 1 </span><span>user</span><span>.</span><span>visit_count</span> <span>+=</span> <span>1</span> <span># Correct approach: Note the capital U, that is, using the attribute of the model, and the generated SQL is to add 1 on the SQL server side # Corresponding SQL: Update users set visit_count = visit_count + 1 where user.id = 1 </span><span>user</span><span>.</span><span>visit_count</span> <span>=</span> <span>User</span><span>.</span><span>visit_count</span> <span>+</span> <span>1</span> <span># To add an object, directly use the session.add method </span><span>session</span><span>.</span><span>add</span><span>(</span><span>user</span><span>)</span> <span># Or add_all </span><span>session</span><span>.</span><span>add_all</span><span>([</span><span>user1</span><span>,</span> <span>user2</span><span>,</span> <span>group1</span><span>])</span> <span># If you want to get the inserted ID, of course, you can also read it after committing </span><span>session</span><span>.</span><span>flush</span><span>()</span> <span># flush is not a commit, and the transaction has not been committed. It should be repeatable read, which is related to the isolation level of the database. </span><span>print</span><span>(</span><span>user</span><span>.</span><span>id</span><span>)</span> <span># To delete, use session.delete </span><span>session</span><span>.</span><span>delete</span><span>(</span><span>user</span><span>)</span>from sqlalchemy import select # The parameter of where is an expression composed of `==`. The advantage is that when writing code, spelling errors will be detected stmt = select(User).where(User.name == "john").order_by(User.id) # filter_by uses **kwargs as parameters stmt = select(User).filter_by(name="some_user") # order_by can also use User.id.desc() to represent reverse sorting result = session.execute(stmt) # Generally, when selecting the entire object, the scalars method should be used, otherwise a tuple containing one object will be returned for user in result.scalars(): print(user.name) # When querying a single attribute of the model, there is no need to use scalars result = session.execute(select(User.name)) for row in result: print(row.name) # There is also a shortcut to query by id: user = session.get(User, pk=1) # To update data, the update statement needs to be used from sqlalchemy import update # synchronize_session has three options: false, "fetch", "evaluate", and the default is evaluate # false means not updating the object in Python at all # fetch means reloading an object from the database # evaluate means that while updating the database, the same operation is also tried on the object in Python as much as possible stmt = update(User).where(User.name == "john").values(name="John").execution_options(synchronize_session="fetch") session.execute(stmt) # Or directly assign a value to the attribute user.name = "John" session.commit() # There is a place here that may introduce a race condition (竞态条件) # Wrong! If two processes update this value at the same time, it may result in only one value being updated. # Both assign the value they think is correct, which is 2, but the actual correct value is 1 + 1 + 1 = 3 # Corresponding SQL: Update users set visit_count = 2 where user.id = 1 user.visit_count += 1 # Correct approach: Note the capital U, that is, using the attribute of the model, and the generated SQL is to add 1 on the SQL server side # Corresponding SQL: Update users set visit_count = visit_count + 1 where user.id = 1 user.visit_count = User.visit_count + 1 # To add an object, directly use the session.add method session.add(user) # Or add_all session.add_all([user1, user2, group1]) # If you want to get the inserted ID, of course, you can also read it after committing session.flush() # flush is not a commit, and the transaction has not been committed. It should be repeatable read, which is related to the isolation level of the database. print(user.id) # To delete, use session.delete session.delete(user)
Enter fullscreen mode Exit fullscreen mode
Loading Associated Models
If after reading a list of N records, you then go to the database to read the specific values of each item one by one, N+1 queries will be generated, which is the most common mistake in the database: the N+1 problem.
By default, the foreign key associated models will not be loaded in the query. You can use the selectinload option to load the foreign keys, thus avoiding the N+1 problem.
<span># Foreign keys not loaded </span><span>session</span><span>.</span><span>execute</span><span>(</span><span>select</span><span>(</span><span>User</span><span>)).</span><span>scalars</span><span>().</span><span>all</span><span>()</span><span># Foreign keys loaded </span><span>session</span><span>.</span><span>execute</span><span>(</span><span>select</span><span>(</span><span>User</span><span>).</span><span>options</span><span>(</span><span>selectinload</span><span>(</span><span>User</span><span>.</span><span>groups</span><span>))).</span><span>scalars</span><span>().</span><span>all</span><span>()</span><span># Foreign keys not loaded </span><span>session</span><span>.</span><span>execute</span><span>(</span><span>select</span><span>(</span><span>User</span><span>)).</span><span>scalars</span><span>().</span><span>all</span><span>()</span> <span># Foreign keys loaded </span><span>session</span><span>.</span><span>execute</span><span>(</span><span>select</span><span>(</span><span>User</span><span>).</span><span>options</span><span>(</span><span>selectinload</span><span>(</span><span>User</span><span>.</span><span>groups</span><span>))).</span><span>scalars</span><span>().</span><span>all</span><span>()</span># Foreign keys not loaded session.execute(select(User)).scalars().all() # Foreign keys loaded session.execute(select(User).options(selectinload(User.groups))).scalars().all()
Enter fullscreen mode Exit fullscreen mode
The principle of Selectinload is to use the select in subquery. In addition to selectinload, the traditional joinedload can also be used, and its principle is the most common join table.
<span># Use joinedload to load foreign keys. Note that the unique method needs to be used, which is specified in 2.0. </span><span>session</span><span>.</span><span>execute</span><span>(</span><span>select</span><span>(</span><span>User</span><span>).</span><span>options</span><span>(</span><span>joinedload</span><span>(</span><span>User</span><span>.</span><span>groups</span><span>))).</span><span>unique</span><span>().</span><span>scalars</span><span>().</span><span>all</span><span>()</span><span># Use joinedload to load foreign keys. Note that the unique method needs to be used, which is specified in 2.0. </span><span>session</span><span>.</span><span>execute</span><span>(</span><span>select</span><span>(</span><span>User</span><span>).</span><span>options</span><span>(</span><span>joinedload</span><span>(</span><span>User</span><span>.</span><span>groups</span><span>))).</span><span>unique</span><span>().</span><span>scalars</span><span>().</span><span>all</span><span>()</span># Use joinedload to load foreign keys. Note that the unique method needs to be used, which is specified in 2.0. session.execute(select(User).options(joinedload(User.groups))).unique().scalars().all()
Enter fullscreen mode Exit fullscreen mode
In 2.0, it is more recommended to use selectinload rather than joinedload. Generally, selectinload is better, and there is no need to use unique.
Writing Foreign Keys
In SQLAlchemy, you can directly handle foreign keys just like handling arrays.
<span>user</span><span>.</span><span>permissions</span><span>.</span><span>append</span><span>(</span><span>open_permission</span><span>)</span> <span># Add </span><span>user</span><span>.</span><span>permissions</span><span>.</span><span>remove</span><span>(</span><span>save_permission</span><span>)</span> <span># Remove # Clear all foreign keys </span><span>user</span><span>.</span><span>permissions</span><span>.</span><span>clear</span><span>()</span><span>user</span><span>.</span><span>permissions</span> <span>=</span> <span>[]</span><span>user</span><span>.</span><span>permissions</span><span>.</span><span>append</span><span>(</span><span>open_permission</span><span>)</span> <span># Add </span><span>user</span><span>.</span><span>permissions</span><span>.</span><span>remove</span><span>(</span><span>save_permission</span><span>)</span> <span># Remove # Clear all foreign keys </span><span>user</span><span>.</span><span>permissions</span><span>.</span><span>clear</span><span>()</span> <span>user</span><span>.</span><span>permissions</span> <span>=</span> <span>[]</span>user.permissions.append(open_permission) # Add user.permissions.remove(save_permission) # Remove # Clear all foreign keys user.permissions.clear() user.permissions = []
Enter fullscreen mode Exit fullscreen mode
Special Handling of JSON Fields
Most databases now support JSON fields. In SQLAlchemy, you can directly read a JSON object from a field or write a JSON object to it. But never directly perform an update on this JSON object and expect to write it back to the database, which is unreliable. Be sure to copy, read and write, and then assign it back.
<span>import</span> <span>copy</span><span>article</span> <span>=</span> <span>session</span><span>.</span><span>get</span><span>(</span><span>Article</span><span>,</span> <span>1</span><span>)</span><span>tags</span> <span>=</span> <span>copy</span><span>.</span><span>copy</span><span>(</span><span>article</span><span>.</span><span>tags</span><span>)</span><span>tags</span><span>.</span><span>append</span><span>(</span><span>"</span><span>iOS</span><span>"</span><span>)</span><span>article</span><span>.</span><span>tags</span> <span>=</span> <span>tags</span><span>session</span><span>.</span><span>commit</span><span>()</span><span>import</span> <span>copy</span> <span>article</span> <span>=</span> <span>session</span><span>.</span><span>get</span><span>(</span><span>Article</span><span>,</span> <span>1</span><span>)</span> <span>tags</span> <span>=</span> <span>copy</span><span>.</span><span>copy</span><span>(</span><span>article</span><span>.</span><span>tags</span><span>)</span> <span>tags</span><span>.</span><span>append</span><span>(</span><span>"</span><span>iOS</span><span>"</span><span>)</span> <span>article</span><span>.</span><span>tags</span> <span>=</span> <span>tags</span> <span>session</span><span>.</span><span>commit</span><span>()</span>import copy article = session.get(Article, 1) tags = copy.copy(article.tags) tags.append("iOS") article.tags = tags session.commit()
Enter fullscreen mode Exit fullscreen mode
Batch Insertion
When a large amount of data needs to be inserted, if the method of inserting one by one is used, a lot of time will be wasted in the interaction with the database, and the efficiency is very low. Most databases such as MySQL provide the insert ... values (...), (...) ... batch insertion API, and this can also be well utilized in SQLAlchemy.
<span># Use session.bulk_save_objects(...) to directly insert multiple objects </span><span>from</span> <span>sqlalchemy.orm</span> <span>import</span> <span>Session</span><span>s</span> <span>=</span> <span>Session</span><span>()</span><span>objects</span> <span>=</span> <span>[</span><span>User</span><span>(</span><span>name</span><span>=</span><span>"</span><span>u1</span><span>"</span><span>),</span><span>User</span><span>(</span><span>name</span><span>=</span><span>"</span><span>u2</span><span>"</span><span>),</span><span>User</span><span>(</span><span>name</span><span>=</span><span>"</span><span>u3</span><span>"</span><span>)</span><span>]</span><span>s</span><span>.</span><span>bulk_save_objects</span><span>(</span><span>objects</span><span>)</span><span>s</span><span>.</span><span>commit</span><span>()</span><span># Using bulk_insert_mappings can save the overhead of creating objects and directly insert dictionaries </span><span>users</span> <span>=</span> <span>[</span><span>{</span><span>"</span><span>name</span><span>"</span><span>:</span> <span>"</span><span>u1</span><span>"</span><span>},</span><span>{</span><span>"</span><span>name</span><span>"</span><span>:</span> <span>"</span><span>u2</span><span>"</span><span>},</span><span>{</span><span>"</span><span>name</span><span>"</span><span>:</span> <span>"</span><span>u3</span><span>"</span><span>},</span><span>]</span><span>s</span><span>.</span><span>bulk_insert_mappings</span><span>(</span><span>User</span><span>,</span> <span>users</span><span>)</span><span>s</span><span>.</span><span>commit</span><span>()</span><span># Using bulk_update_mappings can update objects in batches. The id in the dictionary will be used as the where condition, # and all other fields will be used for the update </span><span>session</span><span>.</span><span>bulk_update_mappings</span><span>(</span><span>User</span><span>,</span> <span>users</span><span>)</span><span># Use session.bulk_save_objects(...) to directly insert multiple objects </span><span>from</span> <span>sqlalchemy.orm</span> <span>import</span> <span>Session</span> <span>s</span> <span>=</span> <span>Session</span><span>()</span> <span>objects</span> <span>=</span> <span>[</span> <span>User</span><span>(</span><span>name</span><span>=</span><span>"</span><span>u1</span><span>"</span><span>),</span> <span>User</span><span>(</span><span>name</span><span>=</span><span>"</span><span>u2</span><span>"</span><span>),</span> <span>User</span><span>(</span><span>name</span><span>=</span><span>"</span><span>u3</span><span>"</span><span>)</span> <span>]</span> <span>s</span><span>.</span><span>bulk_save_objects</span><span>(</span><span>objects</span><span>)</span> <span>s</span><span>.</span><span>commit</span><span>()</span> <span># Using bulk_insert_mappings can save the overhead of creating objects and directly insert dictionaries </span><span>users</span> <span>=</span> <span>[</span> <span>{</span><span>"</span><span>name</span><span>"</span><span>:</span> <span>"</span><span>u1</span><span>"</span><span>},</span> <span>{</span><span>"</span><span>name</span><span>"</span><span>:</span> <span>"</span><span>u2</span><span>"</span><span>},</span> <span>{</span><span>"</span><span>name</span><span>"</span><span>:</span> <span>"</span><span>u3</span><span>"</span><span>},</span> <span>]</span> <span>s</span><span>.</span><span>bulk_insert_mappings</span><span>(</span><span>User</span><span>,</span> <span>users</span><span>)</span> <span>s</span><span>.</span><span>commit</span><span>()</span> <span># Using bulk_update_mappings can update objects in batches. The id in the dictionary will be used as the where condition, # and all other fields will be used for the update </span><span>session</span><span>.</span><span>bulk_update_mappings</span><span>(</span><span>User</span><span>,</span> <span>users</span><span>)</span># Use session.bulk_save_objects(...) to directly insert multiple objects from sqlalchemy.orm import Session s = Session() objects = [ User(name="u1"), User(name="u2"), User(name="u3") ] s.bulk_save_objects(objects) s.commit() # Using bulk_insert_mappings can save the overhead of creating objects and directly insert dictionaries users = [ {"name": "u1"}, {"name": "u2"}, {"name": "u3"}, ] s.bulk_insert_mappings(User, users) s.commit() # Using bulk_update_mappings can update objects in batches. The id in the dictionary will be used as the where condition, # and all other fields will be used for the update session.bulk_update_mappings(User, users)
Enter fullscreen mode Exit fullscreen mode
DeclarativeBase
Fully embrace the Python native type system
<span>from</span> <span>sqlalchemy.orm</span> <span>import</span> <span>DeclarativeBase</span><span>class</span> <span>Base</span><span>(</span><span>DeclarativeBase</span><span>):</span><span>pass</span><span>from</span> <span>sqlalchemy.orm</span> <span>import</span> <span>mapped_column</span><span>,</span> <span>MappedColumn</span><span>id</span><span>:</span> <span>Mapped</span><span>[</span><span>int</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>Integer</span><span>,</span> <span>primary_key</span><span>=</span><span>True</span><span>)</span><span>fullname</span><span>:</span> <span>Mapped</span><span>[</span><span>Optional</span><span>[</span><span>str</span><span>]]</span><span>from</span> <span>sqlalchemy.orm</span> <span>import</span> <span>DeclarativeBase</span> <span>class</span> <span>Base</span><span>(</span><span>DeclarativeBase</span><span>):</span> <span>pass</span> <span>from</span> <span>sqlalchemy.orm</span> <span>import</span> <span>mapped_column</span><span>,</span> <span>MappedColumn</span> <span>id</span><span>:</span> <span>Mapped</span><span>[</span><span>int</span><span>]</span> <span>=</span> <span>mapped_column</span><span>(</span><span>Integer</span><span>,</span> <span>primary_key</span><span>=</span><span>True</span><span>)</span> <span>fullname</span><span>:</span> <span>Mapped</span><span>[</span><span>Optional</span><span>[</span><span>str</span><span>]]</span>from sqlalchemy.orm import DeclarativeBase class Base(DeclarativeBase): pass from sqlalchemy.orm import mapped_column, MappedColumn id: Mapped[int] = mapped_column(Integer, primary_key=True) fullname: Mapped[Optional[str]]
Enter fullscreen mode Exit fullscreen mode
Asyncio
One AsyncSession per task. The AsyncSession object is a mutable, stateful object that represents an ongoing single, stateful database transaction. When using asyncio for concurrent tasks, such as using APIs like asyncio.gather(), each individual task should use a separate AsyncSession.
<span>from</span> <span>sqlalchemy.ext.asyncio</span> <span>import</span> <span>create_async_engine</span><span>,</span> <span>async_sessionmaker</span><span>,</span> <span>AsyncSession</span><span>engine</span> <span>=</span> <span>create_async_engine</span><span>(</span><span>url</span><span>,</span> <span>echo</span><span>=</span><span>True</span><span>)</span><span>session</span> <span>=</span> <span>async_sessionmaker</span><span>(</span><span>engine</span><span>)</span><span># Create objects </span><span>async</span> <span>with</span> <span>engine</span><span>.</span><span>begin</span><span>()</span> <span>as</span> <span>conn</span><span>:</span><span>await</span> <span>conn</span><span>.</span><span>run_sync</span><span>(</span><span>Base</span><span>.</span><span>metadata</span><span>.</span><span>create_all</span><span>)</span><span># Insert data </span><span>async</span> <span>with</span> <span>session</span><span>()</span> <span>as</span> <span>db</span><span>:</span><span>db</span><span>.</span><span>add</span><span>(...)</span><span>await</span> <span>db</span><span>.</span><span>commit</span><span>()</span><span># Query data </span><span>async</span> <span>with</span> <span>session</span><span>()</span> <span>as</span> <span>db</span><span>:</span><span>stmt</span> <span>=</span> <span>select</span><span>(</span><span>A</span><span>)</span><span>row</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>execute</span><span>(</span><span>stmt</span><span>)</span><span>for</span> <span>obj</span> <span>in</span> <span>row</span><span>.</span><span>scalars</span><span>():</span><span>print</span><span>(</span><span>obj</span><span>.</span><span>id</span><span>)</span><span>await</span> <span>engine</span><span>.</span><span>dispose</span><span>()</span><span>from</span> <span>sqlalchemy.ext.asyncio</span> <span>import</span> <span>create_async_engine</span><span>,</span> <span>async_sessionmaker</span><span>,</span> <span>AsyncSession</span> <span>engine</span> <span>=</span> <span>create_async_engine</span><span>(</span><span>url</span><span>,</span> <span>echo</span><span>=</span><span>True</span><span>)</span> <span>session</span> <span>=</span> <span>async_sessionmaker</span><span>(</span><span>engine</span><span>)</span> <span># Create objects </span><span>async</span> <span>with</span> <span>engine</span><span>.</span><span>begin</span><span>()</span> <span>as</span> <span>conn</span><span>:</span> <span>await</span> <span>conn</span><span>.</span><span>run_sync</span><span>(</span><span>Base</span><span>.</span><span>metadata</span><span>.</span><span>create_all</span><span>)</span> <span># Insert data </span><span>async</span> <span>with</span> <span>session</span><span>()</span> <span>as</span> <span>db</span><span>:</span> <span>db</span><span>.</span><span>add</span><span>(...)</span> <span>await</span> <span>db</span><span>.</span><span>commit</span><span>()</span> <span># Query data </span><span>async</span> <span>with</span> <span>session</span><span>()</span> <span>as</span> <span>db</span><span>:</span> <span>stmt</span> <span>=</span> <span>select</span><span>(</span><span>A</span><span>)</span> <span>row</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>execute</span><span>(</span><span>stmt</span><span>)</span> <span>for</span> <span>obj</span> <span>in</span> <span>row</span><span>.</span><span>scalars</span><span>():</span> <span>print</span><span>(</span><span>obj</span><span>.</span><span>id</span><span>)</span> <span>await</span> <span>engine</span><span>.</span><span>dispose</span><span>()</span>from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession engine = create_async_engine(url, echo=True) session = async_sessionmaker(engine) # Create objects async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) # Insert data async with session() as db: db.add(...) await db.commit() # Query data async with session() as db: stmt = select(A) row = await db.execute(stmt) for obj in row.scalars(): print(obj.id) await engine.dispose()
Enter fullscreen mode Exit fullscreen mode
Using in a Multiprocessing Environment
Due to the Global Interpreter Lock (GIL) in Python, to utilize multi-core processors, multiprocessing needs to be used. In a multiprocessing environment, resources cannot be shared. Corresponding to SQLAlchemy, that is, the connection pool cannot be shared. We need to solve this problem manually.
Generally speaking, it is best not to try to share the same Session among multiple processes. It is best to create a Session when initializing each process.
Adding Where Conditions Only When a Value is Set
In the URL, it is often necessary to return corresponding results according to which options the user has specified.
<span>query</span> <span>=</span> <span>select</span><span>(</span><span>User</span><span>)</span><span>if</span> <span>username</span> <span>is</span> <span>not</span> <span>None</span><span>:</span><span>query</span> <span>=</span> <span>query</span><span>.</span><span>where</span><span>(</span><span>User</span><span>.</span><span>username</span> <span>==</span> <span>username</span><span>)</span><span>if</span> <span>password</span> <span>is</span> <span>not</span> <span>None</span><span>:</span><span>query</span> <span>=</span> <span>query</span><span>.</span><span>where</span><span>(</span><span>User</span><span>.</span><span>password</span> <span>==</span> <span>password</span><span>)</span><span>query</span> <span>=</span> <span>select</span><span>(</span><span>User</span><span>)</span> <span>if</span> <span>username</span> <span>is</span> <span>not</span> <span>None</span><span>:</span> <span>query</span> <span>=</span> <span>query</span><span>.</span><span>where</span><span>(</span><span>User</span><span>.</span><span>username</span> <span>==</span> <span>username</span><span>)</span> <span>if</span> <span>password</span> <span>is</span> <span>not</span> <span>None</span><span>:</span> <span>query</span> <span>=</span> <span>query</span><span>.</span><span>where</span><span>(</span><span>User</span><span>.</span><span>password</span> <span>==</span> <span>password</span><span>)</span>query = select(User) if username is not None: query = query.where(User.username == username) if password is not None: query = query.where(User.password == password)
Enter fullscreen mode Exit fullscreen mode
Leapcell: The Next-Gen Serverless Platform for Python app Hosting
Finally, I would like to recommend to you the platform that is most suitable for deploying Python services: Leapcell
1. Multi-Language Support
- Develop with JavaScript, Python, Go, or Rust.
2. Deploy unlimited projects for free
- pay only for usage — no requests, no charges.
3. Unbeatable Cost Efficiency
- Pay-as-you-go with no idle charges.
- Example: $25 supports 6.94M requests at a 60ms average response time.
4. Streamlined Developer Experience
- Intuitive UI for effortless setup.
- Fully automated CI/CD pipelines and GitOps integration.
- Real-time metrics and logging for actionable insights.
5. Effortless Scalability and High Performance
- Auto-scaling to handle high concurrency with ease.
- Zero operational overhead — just focus on building.
Explore more in the documentation!
Leapcell Twitter: https://x.com/LeapcellHQ
© 版权声明
THE END
暂无评论内容