diff --git a/docs/codeql/support/reusables/frameworks.rst b/docs/codeql/support/reusables/frameworks.rst
index b186c08c5d6..52e96f9af0c 100644
--- a/docs/codeql/support/reusables/frameworks.rst
+++ b/docs/codeql/support/reusables/frameworks.rst
@@ -183,6 +183,7 @@ Python built-in support
    pydantic, Utility library
    yarl, Utility library
    aioch, Database
+   aiopg, Database
    asyncpg, Database
    clickhouse-driver, Database
    mysql-connector-python, Database
diff --git a/python/change-notes/2021-11-09-model-aiopg.md b/python/change-notes/2021-11-09-model-aiopg.md
new file mode 100644
index 00000000000..7bf78a8de01
--- /dev/null
+++ b/python/change-notes/2021-11-09-model-aiopg.md
@@ -0,0 +1,2 @@
+lgtm,codescanning
+* Added modeling of `aiopg` for sinks executing SQL.
diff --git a/python/ql/lib/semmle/python/Frameworks.qll b/python/ql/lib/semmle/python/Frameworks.qll
index 932d2279a5a..b0587f9d430 100644
--- a/python/ql/lib/semmle/python/Frameworks.qll
+++ b/python/ql/lib/semmle/python/Frameworks.qll
@@ -6,6 +6,7 @@
 // `docs/codeql/support/reusables/frameworks.rst`
 private import semmle.python.frameworks.Aioch
 private import semmle.python.frameworks.Aiohttp
+private import semmle.python.frameworks.Aiopg
 private import semmle.python.frameworks.Asyncpg
 private import semmle.python.frameworks.ClickhouseDriver
 private import semmle.python.frameworks.Cryptodome
diff --git a/python/ql/lib/semmle/python/frameworks/Aiopg.qll b/python/ql/lib/semmle/python/frameworks/Aiopg.qll
new file mode 100644
index 00000000000..927ae90ed05
--- /dev/null
+++ b/python/ql/lib/semmle/python/frameworks/Aiopg.qll
@@ -0,0 +1,141 @@
+/**
+ * Provides classes modeling security-relevant aspects of the `aiopg` PyPI package.
+ * See
+ * - https://aiopg.readthedocs.io/en/stable/index.html
+ * - https://pypi.org/project/aiopg/
+ */
+
+private import python
+private import semmle.python.dataflow.new.DataFlow
+private import semmle.python.Concepts
+private import semmle.python.ApiGraphs
+
+/** Provides models for the `aiopg` PyPI package. */
+private module Aiopg {
+  private import semmle.python.internal.Awaited
+
+  /**
+   * A `ConectionPool` is created when the result of `aiopg.create_pool()` is awaited.
+   * See https://aiopg.readthedocs.io/en/stable/core.html#pool
+   */
+  API::Node connectionPool() {
+    result = API::moduleImport("aiopg").getMember("create_pool").getReturn().getAwaited()
+  }
+
+  /**
+   * A `Connection` is created when
+   * - the result of `aiopg.connect()` is awaited.
+   * - the result of calling `aquire` on a `ConnectionPool` is awaited.
+   * See https://aiopg.readthedocs.io/en/stable/core.html#connection
+   */
+  API::Node connection() {
+    result = API::moduleImport("aiopg").getMember("connect").getReturn().getAwaited()
+    or
+    result = connectionPool().getMember("acquire").getReturn().getAwaited()
+  }
+
+  /**
+   * A `Cursor` is created when
+   * - the result of calling `cursor` on a `ConnectionPool` is awaited.
+   * - the result of calling `cursor` on a `Connection` is awaited.
+   * See https://aiopg.readthedocs.io/en/stable/core.html#cursor
+   */
+  API::Node cursor() {
+    result = connectionPool().getMember("cursor").getReturn().getAwaited()
+    or
+    result = connection().getMember("cursor").getReturn().getAwaited()
+  }
+
+  /**
+   * Calling `execute` on a `Cursor` constructs a query.
+   * See https://aiopg.readthedocs.io/en/stable/core.html#aiopg.Cursor.execute
+   */
+  class CursorExecuteCall extends SqlConstruction::Range, DataFlow::CallCfgNode {
+    CursorExecuteCall() { this = cursor().getMember("execute").getACall() }
+
+    override DataFlow::Node getSql() { result in [this.getArg(0), this.getArgByName("operation")] }
+  }
+
+  /**
+   * This is only needed to connect the argument to the execute call with the subsequnt awaiting.
+   * It should be obsolete once we have `API::CallNode` available.
+   */
+  private DataFlow::TypeTrackingNode cursorExecuteCall(DataFlow::TypeTracker t, DataFlow::Node sql) {
+    // cursor created from connection
+    t.start() and
+    sql = result.(CursorExecuteCall).getSql()
+    or
+    exists(DataFlow::TypeTracker t2 | result = cursorExecuteCall(t2, sql).track(t2, t))
+  }
+
+  DataFlow::Node cursorExecuteCall(DataFlow::Node sql) {
+    cursorExecuteCall(DataFlow::TypeTracker::end(), sql).flowsTo(result)
+  }
+
+  /**
+   * Awaiting the result of calling `execute` executes the query.
+   * See https://aiopg.readthedocs.io/en/stable/core.html#aiopg.Cursor.execute
+   */
+  class AwaitedCursorExecuteCall extends SqlExecution::Range {
+    DataFlow::Node sql;
+
+    AwaitedCursorExecuteCall() { this = awaited(cursorExecuteCall(sql)) }
+
+    override DataFlow::Node getSql() { result = sql }
+  }
+
+  /**
+   * An `Engine` is created when the result of calling `aiopg.sa.create_engine` is awaited.
+   * See https://aiopg.readthedocs.io/en/stable/sa.html#engine
+   */
+  API::Node engine() {
+    result =
+      API::moduleImport("aiopg").getMember("sa").getMember("create_engine").getReturn().getAwaited()
+  }
+
+  /**
+   * A `SAConnection` is created when the result of calling `aquire` on an `Engine` is awaited.
+   * See https://aiopg.readthedocs.io/en/stable/sa.html#connection
+   */
+  API::Node saConnection() { result = engine().getMember("acquire").getReturn().getAwaited() }
+
+  /**
+   * Calling `execute` on a `SAConnection` constructs a query.
+   * See https://aiopg.readthedocs.io/en/stable/sa.html#aiopg.sa.SAConnection.execute
+   */
+  class SAConnectionExecuteCall extends SqlConstruction::Range, DataFlow::CallCfgNode {
+    SAConnectionExecuteCall() { this = saConnection().getMember("execute").getACall() }
+
+    override DataFlow::Node getSql() { result in [this.getArg(0), this.getArgByName("query")] }
+  }
+
+  /**
+   * This is only needed to connect the argument to the execute call with the subsequnt awaiting.
+   * It should be obsolete once we have `API::CallNode` available.
+   */
+  private DataFlow::TypeTrackingNode saConnectionExecuteCall(
+    DataFlow::TypeTracker t, DataFlow::Node sql
+  ) {
+    // saConnection created from engine
+    t.start() and
+    sql = result.(SAConnectionExecuteCall).getSql()
+    or
+    exists(DataFlow::TypeTracker t2 | result = saConnectionExecuteCall(t2, sql).track(t2, t))
+  }
+
+  DataFlow::Node saConnectionExecuteCall(DataFlow::Node sql) {
+    saConnectionExecuteCall(DataFlow::TypeTracker::end(), sql).flowsTo(result)
+  }
+
+  /**
+   * Awaiting the result of calling `execute` executes the query.
+   * See https://aiopg.readthedocs.io/en/stable/sa.html#aiopg.sa.SAConnection.execute
+   */
+  class AwaitedSAConnectionExecuteCall extends SqlExecution::Range {
+    DataFlow::Node sql;
+
+    AwaitedSAConnectionExecuteCall() { this = awaited(saConnectionExecuteCall(sql)) }
+
+    override DataFlow::Node getSql() { result = sql }
+  }
+}
diff --git a/python/ql/test/library-tests/frameworks/aiopg/ConceptsTest.expected b/python/ql/test/library-tests/frameworks/aiopg/ConceptsTest.expected
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/python/ql/test/library-tests/frameworks/aiopg/ConceptsTest.ql b/python/ql/test/library-tests/frameworks/aiopg/ConceptsTest.ql
new file mode 100644
index 00000000000..b557a0bccb6
--- /dev/null
+++ b/python/ql/test/library-tests/frameworks/aiopg/ConceptsTest.ql
@@ -0,0 +1,2 @@
+import python
+import experimental.meta.ConceptsTest
diff --git a/python/ql/test/library-tests/frameworks/aiopg/test.py b/python/ql/test/library-tests/frameworks/aiopg/test.py
new file mode 100644
index 00000000000..63bf141f52d
--- /dev/null
+++ b/python/ql/test/library-tests/frameworks/aiopg/test.py
@@ -0,0 +1,33 @@
+import aiopg
+
+# Only a cursor can execute sql.
+async def test_cursor():
+    # Create connection directly
+    conn = await aiopg.connect()
+    cur = await conn.cursor()
+    await cur.execute("sql")  # $ getSql="sql" constructedSql="sql"
+
+    # Create connection via pool
+    async with aiopg.create_pool() as pool:
+        # Create Cursor via Connection
+        async with pool.acquire() as conn:
+            async with conn.cursor() as cur:
+                await cur.execute("sql")  # $ getSql="sql" constructedSql="sql"
+
+        # Create Cursor directly
+        async with pool.cursor() as cur:
+            await cur.execute("sql")  # $ getSql="sql" constructedSql="sql"
+
+    # variants using as few `async with` as possible
+    pool = await aiopg.create_pool()
+    conn = await pool.acquire()
+    cur = await conn.cursor()
+    await cur.execute("sql")  # $ getSql="sql" constructedSql="sql"
+
+# Test SQLAlchemy integration
+from aiopg.sa import create_engine
+
+async def test_engine():
+    engine = await create_engine()
+    conn = await engine.acquire()
+    await conn.execute("sql")  # $ getSql="sql" constructedSql="sql"