Skip to content

New method Env.set_api_key. #40

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Oct 7, 2025
Merged

New method Env.set_api_key. #40

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions CHANGES.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,14 @@ Changelog
---------


2.x.x (unreleased)
~~~~~~~~~~~~~~~~~~

* Fix :meth:`Client.clone_database`.

* New method :meth:`Env.set_api_key`.


2.4.1 (2025-10-06)
~~~~~~~~~~~~~~~~~~

Expand Down
2 changes: 2 additions & 0 deletions docs/api.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -190,6 +190,8 @@ Environment

.. automethod:: generate_api_key

.. automethod:: set_api_key

.. automethod:: session_authenticate

.. automethod:: session_destroy
Expand Down
54 changes: 32 additions & 22 deletions odooly.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@
except ImportError:
requests = None

__version__ = '2.4.1'
__version__ = '2.4.1.dev0'
__all__ = ['Client', 'Env', 'WebAPI', 'Service', 'Json2',
'Printer', 'Error', 'ServerError',
'BaseModel', 'Model', 'BaseRecord', 'Record', 'RecordList',
Expand Down Expand Up @@ -695,46 +695,40 @@ def _auth(self, user, password, api_key):
info['user_context'] = user_context or {}
return (uid, password, info)

def _set_credentials(self, uid, api_key, store_api_key=True):
def set_api_key(self, api_key, store=True):
"""Configure methods to use an API key."""
def env_auth(method): # Authenticated endpoints
return partial(method, self.db_name, uid, api_key)
return partial(method, self.db_name, self.uid, api_key)
if self.client.web and self.client.version_info >= 19.0:
self._json2 = Json2(self.client, self.db_name, api_key)._check()
if self.client._object:
if self.client._object: # RPC endpoint if available
self._execute = env_auth(self.client._object.execute)
self._execute_kw = env_auth(self.client._object.execute_kw)
else: # WebAPI and JSON-2
else: # Otherwise, use JSON-2 or WebAPI
self._execute_kw = self._json2 or self._call_kw
if self.client._report: # Odoo <= 10
if store:
self._api_key = api_key

if self.client._report: # Odoo < 11
self.exec_workflow = env_auth(self.client._object.exec_workflow)
self.report = env_auth(self.client._report.report)
self.report_get = env_auth(self.client._report.report_get)
self.render_report = env_auth(self.client._report.render_report)
if self.client._wizard: # OpenERP 6.1
self.wizard_execute = env_auth(self.client._wizard.execute)
self.wizard_create = env_auth(self.client._wizard.create)
if store_api_key:
self._api_key = api_key
return api_key

def _configure(self, uid, user, password, api_key, context, session):
need_auth = uid != self.uid or (api_key and api_key != self._api_key)
if self.uid: # Create a new Env() instance
env = Env(self.client)
(env.db_name, env.name) = (self.db_name, self.name)
env._model_names = self._model_names
env._models = {}
else: # Configure the Env() instance
env = self
if uid == self.uid: # Copy methods
for key in ('_execute', '_execute_kw', 'exec_workflow',
'report', 'report_get', 'render_report',
'wizard_execute', 'wizard_create'):
if hasattr(self, key):
setattr(env, key, getattr(self, key))
env._api_key = self._api_key
env._json2 = self._json2
else: # Create methods
env._set_credentials(uid, api_key or password, bool(api_key))

# Setup uid and user
if isinstance(user, Record):
user = user.login
Expand All @@ -746,6 +740,18 @@ def _configure(self, uid, user, password, api_key, context, session):
assert isinstance(user, str), repr(user)
env.user.__dict__['login'] = user
env.user._cached_keys.add('login')

# Set API methods
if need_auth:
env.set_api_key(api_key or password, bool(api_key))
else: # Copy methods
for key in '_execute_kw', '_api_key', '_json2':
setattr(env, key, getattr(self, key))
for key in ('_execute', 'exec_workflow',
'report', 'report_get', 'render_report',
'wizard_execute', 'wizard_create'):
if hasattr(self, key):
setattr(env, key, getattr(self, key))
return env

@property
Expand Down Expand Up @@ -1109,7 +1115,7 @@ def generate_api_key(self):
res = wiz.make_key()
self.user.refresh()
assert res['res_model'] == "res.users.apikeys.show"
return self._set_credentials(self.uid, res['context']['default_key'])
return self.set_api_key(res['context']['default_key'])


class Client:
Expand Down Expand Up @@ -1453,11 +1459,15 @@ def clone_database(self, passwd, database, neutralize_database=False):
new_name=database, **extra)
# Copy the cache for authentication
auth_cache = self.env._cache_get('auth')
self.env._cache_set('auth', dict(auth_cache), db_name=database)
self.env._cache_set('auth', {**auth_cache}, db_name=database)

# Login with the current user into the new database
(uid, password, __) = self.env._auth(self.env.uid, None, self.env._api_key)
return self.login(self.env.user.login, password, database=database)
auth_args = {
'password': auth_cache[self.env.uid][1],
'api_key': self.env._api_key,
'database': database,
}
return self.login(self.env.user.login, **auth_args)

def drop_database(self, passwd, database):
"""Drop the database.
Expand Down
42 changes: 33 additions & 9 deletions tests/test_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -349,7 +349,7 @@ def obj_exec(self, *args):

def test_create_database(self):
create_database = self.client.create_database
self.client.db.list.side_effect = [['db1'], ['db2']]
self.client.db.list.side_effect = [['db1'], ['db2'], ['db3']]

create_database('abc', 'db1')
create_database('xyz', 'db2', user_password='secret', lang='fr_FR')
Expand All @@ -363,6 +363,11 @@ def test_create_database(self):
call.db.list(),
call.common.login('db2', 'admin', 'secret'),
call.object.execute_kw('db2', self.uid, 'secret', 'res.users', 'context_get', ()),
# Odoo >= 9
call.db.create_database('xyz', 'db3', False, 'fr_FR', 'secret', 'other_login', 'CA'),
call.db.list(),
call.common.login('db3', 'other_login', 'secret'),
call.object.execute_kw('db3', self.uid, 'secret', 'res.users', 'context_get', ()),
]

if float(self.server_version) <= 8.0:
Expand All @@ -371,16 +376,35 @@ def test_create_database(self):
user_password='secret', lang='fr_FR', login='other_login', country_code='CA',
)
self.assertRaises(odooly.Error, create_database, 'xyz', 'db3', login='other_login')
else: # Odoo >= 9.0
self.client.db.list.side_effect = [['db3']]
del expected_calls[-4:]
else: # Odoo >= 9
create_database('xyz', 'db3', user_password='secret', lang='fr_FR', login='other_login', country_code='CA')
self.assertCalls(*expected_calls)
self.assertOutput('')

expected_calls += [
call.db.create_database('xyz', 'db3', False, 'fr_FR', 'secret', 'other_login', 'CA'),
call.db.list(),
call.common.login('db3', 'other_login', 'secret'),
call.object.execute_kw('db3', self.uid, 'secret', 'res.users', 'context_get', ()),
]
def test_clone_database(self):
clone_database = self.client.clone_database
self.client.db.list.side_effect = [['database', 'db1'], ['database', 'db1', 'db2']]

clone_database('abc', 'db1')

expected_calls = [
call.db.duplicate_database('abc', 'database', 'db1'),
call.db.list(),
call.common.login('db1', 'user', 'passwd'),
call.object.execute_kw('db1', 4, 'passwd', 'res.users', 'context_get', ()),
call.db.duplicate_database('xyz', 'db1', 'db2', True),
call.db.list(),
call.common.login('db2', 'user', 'passwd'),
call.object.execute_kw('db2', 4, 'passwd', 'res.users', 'context_get', ()),
]

if float(self.server_version) < 16.0:
# Error: Argument 'neutralize_database' is not supported
self.assertRaises(odooly.Error, clone_database, 'xyz', 'db2', neutralize_database=True)
del expected_calls[-4:]
else:
clone_database('xyz', 'db2', neutralize_database=True)
self.assertCalls(*expected_calls)
self.assertOutput('')

Expand Down