-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest_json_rpc.py
89 lines (75 loc) · 3.49 KB
/
test_json_rpc.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
import json
import random
import urllib.request
# Param
URL = 'http://127.0.18.1:8069'
DB = 'v18c_rpc_helper'
USER = 'admin'
PASS = 'admin'
# See function details in Web Services Odoo documentation
def json_rpc(url, method, params):
data = {
"jsonrpc": "2.0",
"method": method,
"params": params,
"id": random.randint(0, 1000000000),
}
req = urllib.request.Request(url=url, data=json.dumps(data).encode(), headers={
"Content-Type":"application/json",
})
reply = json.loads(urllib.request.urlopen(req).read().decode('UTF-8'))
if reply.get("error"):
raise Exception(reply["error"])
return reply["result"]
def call(url, service, method, *args):
return json_rpc(url, "call", {"service": service, "method": method, "args": args})
# Log in the given database
url = f"{URL}/jsonrpc"
uid = call(url, "common", "login", DB, USER, PASS)
# List Partner records
result_list = call(url, "object", "execute_kw", DB, uid, PASS, 'res.partner', 'search', [[['is_company', '=', True]]])
print("\n--- List Records ---")
print("Id: ", result_list)
# Pagination Partner records
result_pagination = call(url, "object", "execute_kw", DB, uid, PASS, 'res.partner', 'search', [[['is_company', '=', True]]], {'offset': 3, 'limit': 5})
print("\n--- Pagination Records ---")
print("Id: ", result_pagination)
# Count Partner records
result_count = call(url, "object", "execute_kw", DB, uid, PASS, 'res.partner', 'search_count', [[['is_company', '=', True]]])
print("\n--- Count Records ---")
print("Id: ", result_count)
# Read Partner record
result_read = call(url, "object", "execute", DB, uid, PASS, 'res.partner', 'read', [27])
print("\n--- Read Record ---")
print("Id: ", result_read[0]['id'])
print("Name: ", result_read[0]['name'])
# Read Partner with execute_kw method
# execute_kw is an enhanced version of execute that supports both positional arguments and keyword arguments, providing more flexibility.
result_read_kw = call(url, "object", "execute_kw", DB, uid, PASS, 'res.partner', 'read',
[27],
{'fields': ['name', 'email']})
print("\n--- Read Record with kw ---")
print("Result read kw: ", result_read_kw)
# List record fields
result_list_fields = call(url, "object", "execute_kw", DB, uid, PASS, 'res.partner', 'fields_get', [], {'attributes': ['string', 'type']})
print("\n--- List record fields ---")
print("Id: ", result_list_fields)
# Search and read
result_search_read = call(url, "object", "execute_kw", DB, uid, PASS, 'res.partner', 'search_read', [[['is_company', '=', True]]],
{'fields': ['name', 'country_id', 'comment'], 'limit': 5})
print("\n--- Search and read ---")
print("Id: ", result_search_read)
# Create records
# result_create = call(url, "object", "execute_kw", DB, uid, PASS, 'res.partner', 'create', [{'name': "New Partner"}])
# print("\n--- Create records ---")
# print("Id: ", result_create)
# Update records
# result_update = call(url, "object", "execute_kw", DB, uid, PASS, 'res.partner', 'write', [[42], {'name': "Newer partner"}])
# result_update = call(url, "object", "execute_kw", DB, uid, PASS, 'res.partner', 'read', [[42], ['display_name']])
# print("\n--- Update records ---")
# print("Id: ", result_update)
# Delete records
# result_delete = call(url, "object", "execute_kw", DB, uid, PASS, 'res.partner', 'unlink', [[42]])
# result_delete = call(url, "object", "execute_kw", DB, uid, PASS, 'res.partner', 'search', [[['id', '=', 42]]])
# print("\n--- Delete records ---")
# print("Id: ", result_delete)