Skip to content

Commit aa4767e

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

5 files changed

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

0 commit comments

Comments
 (0)