Skip to content

Commit e9cdcea

Browse files
[IMP] fastapi: parse db param in sessionless requests
It is useful for multi-database environments.
1 parent ab196d9 commit e9cdcea

5 files changed

Lines changed: 118 additions & 0 deletions

File tree

fastapi/__init__.py

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
11
from . import models
22
from . import fastapi_dispatcher
33
from . import error_handlers
4+
from . import patch

fastapi/patch.py

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
# Copyright 2026 ForgeFlow S.L. (https://www.forgeflow.com)
2+
# License LGPL-3.0 or later (http://www.gnu.org/licenses/LGPL).
3+
4+
from urllib.parse import parse_qs, urlparse
5+
6+
from odoo.http import Request, db_filter
7+
8+
_original_get_session_and_dbname = Request._get_session_and_dbname
9+
10+
11+
def _patched_get_session_and_dbname(self):
12+
session, dbname = _original_get_session_and_dbname(self)
13+
if not dbname:
14+
# Try to get the database from request "db" param.
15+
# This is useful for multi-database environments.
16+
db = self.httprequest.args.get("db", "").strip()
17+
if not db:
18+
# For browser sub-requests (e.g. openapi.json or static assets fetched
19+
# by Swagger UI / ReDoc), the browser sets a Referer header pointing
20+
# to the docs page which may carry the "db" query parameter.
21+
referer = self.httprequest.headers.get("Referer", "")
22+
if referer:
23+
parsed = urlparse(referer)
24+
db = parse_qs(parsed.query).get("db", [""])[0].strip()
25+
if db and db_filter([db]):
26+
dbname = db
27+
session.db = dbname
28+
return session, dbname
29+
30+
31+
Request._get_session_and_dbname = _patched_get_session_and_dbname

fastapi/readme/CONTRIBUTORS.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1 +1,2 @@
11
- Laurent Mignon \<<laurent.mignon@acsone.eu>\>
2+
- Lois Rilo \<<lois.rilo@forgeflow.com>\>

fastapi/tests/__init__.py

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,2 +1,3 @@
11
from . import test_fastapi
22
from . import test_fastapi_demo
3+
from . import test_db_param

fastapi/tests/test_db_param.py

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,84 @@
1+
# Copyright 2026 ForgeFlow S.L. (https://www.forgeflow.com)
2+
# License LGPL-3.0 or later (http://www.gnu.org/licenses/LGPL).
3+
4+
from unittest.mock import MagicMock, patch
5+
6+
from odoo.http import Request
7+
from odoo.tests import tagged
8+
from odoo.tests.common import TransactionCase
9+
10+
11+
class TestPatchGetSessionAndDbname(TransactionCase):
12+
"""Tests for the _patched_get_session_and_dbname patch.
13+
14+
The patch extends Request._get_session_and_dbname to resolve the database
15+
from a ?db= query parameter or a Referer header when Odoo's normal session
16+
mechanism cannot determine the database (e.g. multi-database staging setups).
17+
"""
18+
19+
def _make_request(self, db_param="", referer=""):
20+
mock_request = MagicMock()
21+
mock_request.httprequest.args = {"db": db_param} if db_param else {}
22+
mock_request.httprequest.headers = {"Referer": referer} if referer else {}
23+
return mock_request
24+
25+
def _call_patched(self, mock_request, original_return, db_filter_return):
26+
with (
27+
patch(
28+
"odoo.addons.fastapi.patch._original_get_session_and_dbname",
29+
return_value=original_return,
30+
),
31+
patch(
32+
"odoo.addons.fastapi.patch.db_filter",
33+
return_value=db_filter_return,
34+
),
35+
):
36+
return Request._get_session_and_dbname(mock_request)
37+
38+
def test_db_already_resolved(self):
39+
"""When the original method resolves a dbname, the patch must not interfere."""
40+
session = MagicMock()
41+
result_session, result_db = self._call_patched(
42+
self._make_request(),
43+
original_return=(session, "existing_db"),
44+
db_filter_return=["existing_db"],
45+
)
46+
self.assertEqual(result_db, "existing_db")
47+
self.assertEqual(result_session, session)
48+
49+
def test_db_from_query_param(self):
50+
"""When dbname is absent, resolve it from the ?db= query parameter."""
51+
session = MagicMock()
52+
result_session, result_db = self._call_patched(
53+
self._make_request(db_param="mydb"),
54+
original_return=(session, None),
55+
db_filter_return=["mydb"],
56+
)
57+
self.assertEqual(result_db, "mydb")
58+
self.assertEqual(session.db, "mydb")
59+
60+
def test_db_from_referer_header(self):
61+
"""When ?db= is absent, fall back to the db param in the Referer header.
62+
63+
This covers browser sub-requests (e.g. openapi.json, Swagger UI assets)
64+
where the Referer points to the docs page carrying ?db=.
65+
"""
66+
session = MagicMock()
67+
result_session, result_db = self._call_patched(
68+
self._make_request(referer="https://example.com/docs?db=mydb"),
69+
original_return=(session, None),
70+
db_filter_return=["mydb"],
71+
)
72+
self.assertEqual(result_db, "mydb")
73+
self.assertEqual(session.db, "mydb")
74+
75+
def test_db_rejected_by_filter(self):
76+
"""A db name that fails db_filter must not be used."""
77+
session = MagicMock()
78+
result_session, result_db = self._call_patched(
79+
self._make_request(db_param="forbidden_db"),
80+
original_return=(session, None),
81+
db_filter_return=[],
82+
)
83+
self.assertIsNone(result_db)
84+
self.assertFalse(hasattr(session, "db") and session.db == "forbidden_db")

0 commit comments

Comments
 (0)