thirdparty
/
CybroAddons
mirror of https://github.com/CybroOdoo/CybroAddons.git
5
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Allow queries by model's fields 

* All requests now can use any fields in the model except for binary
* Fields now support wildcard (*) to output all model fields
* GET request will return ID field if no fields are given
* Return proper HTTP error codes and messages in JSON format by default
* Add pagination support
pull/347/head
drpsyko101 9 months ago
parent
71300f2b48
commit
496f813ac5
1 changed files with 206 additions and 192 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 360
      rest_api_odoo/controllers/rest_api_odoo.py

360
rest_api_odoo/controllers/rest_api_odoo.py
View File

@ -19,12 +19,14 @@
# If not, see <http://www.gnu.org/licenses/>. # If not, see <http://www.gnu.org/licenses/>.
# #
############################################################################# #############################################################################
import json import json
import logging import logging
from datetime import datetime
from datetime import datetime
from odoo import http from odoo import http
from odoo.http import request from odoo.http import request, Response
from ast import literal_eval
_logger = logging.getLogger(__name__) _logger = logging.getLogger(__name__)
@ -36,220 +38,232 @@ class RestApi(http.Controller):
def auth_api_key(self, api_key): def auth_api_key(self, api_key):
"""This function is used to authenticate the api-key when sending a """This function is used to authenticate the api-key when sending a
request""" request"""
user_id = request.env['res.users'].sudo().search([('api_key', '=', api_key)])
user_id = request.env["res.users"].search([("api_key", "=", api_key)])
if api_key is not None and user_id: if api_key is not None and user_id:
response = True return Response(json.dumps({"message": "Authorized"}), status=200)
elif not user_id: elif not user_id:
response = ('<html><body><h2>Invalid <i>API Key</i> ' return Response(json.dumps({"message": "Invalid API Key"}), status=401)
'!</h2></body></html>') return Response(json.dumps({"message": "No API Key Provided"}), status=400)
else:
response = ("<html><body><h2>No <i>API Key</i> Provided " def simplest_type(self, input):
"!</h2></body></html>") """Try cast input into native Python class, otherwise return as string"""
return response try:
return literal_eval(input)
except Exception:
# Handle lowercase booleans
if input == "true":
return True
if input == "false":
return False
return input
def sanitize_records(self, records):
"""Sanitize records for response"""
for record in records:
for key, value in record.items():
# Manually convert datetime fields to string format
if isinstance(value, datetime):
record[key] = value.isoformat()
return records
def generate_response(self, method, model, rec_id): def generate_response(self, method, **query):
"""This function is used to generate the response based on the type """This function is used to generate the response based on the type
of request and the parameters given""" of request and the parameters given"""
option = request.env['connection.api'].search( try:
[('model_id', '=', model)], limit=1) model = query.pop("model")
option = request.env["connection.api"].search(
[("model_id", "=", model)], limit=1
)
model_name = option.model_id.model model_name = option.model_id.model
if method != 'DELETE': model_display_name = option.model_id.name
try:
data = json.loads(request.httprequest.data) data = json.loads(request.httprequest.data)
else: except Exception:
data = {} data = {}
fields = [] fields = []
if data: if data:
for field in data['fields']: for field in data["fields"]:
fields.append(field) fields.append(field)
if not fields and method != 'DELETE':
return ("<html><body><h2>No fields selected for the model" # Return records' ID by default if not specified
"</h2></body></html>") if not fields:
if not option: fields.append("id")
return ("<html><body><h2>No Record Created for the model"
"</h2></body></html>") # Get all model's fields if wildcard is used
try: if "*" in fields:
if method == 'GET':
fields = [] fields = []
for field in data['fields']: record_fields = request.env[str(model_name)].fields_get(
[], attributes=["type"]
)
for field, value in record_fields.items():
value_type = value.get("type")
if not (value_type == "binary"):
fields.append(field) fields.append(field)
if not option:
raise NotImplementedError("No Record Created for the model. ")
if method == "GET":
if not option.is_get: if not option.is_get:
return ("<html><body><h2>Method Not Allowed" raise NameError()
"</h2></body></html>") limit = 0
else: if query.get("limit"):
datas = [] limit = int(str(query.get("limit")))
if rec_id != 0: offset = 0
partner_records = request.env[ if query.get("offset"):
str(model_name) offset = int(str(query.get("offset")))
].search_read(
domain=[('id', '=', rec_id)],
fields=fields
)
# Manually convert datetime fields to string format domains = []
for record in partner_records: for key, value in query.items():
for key, value in record.items(): if not (key == "limit" or key == "offset"):
if isinstance(value, datetime): domains.append((key, "=", self.simplest_type(value)))
record[key] = value.isoformat() partner_records = request.env[str(model_name)].search_read(
data = json.dumps({ domains, fields, limit=limit, offset=offset
'records': partner_records
})
datas.append(data)
return request.make_response(data=datas)
else:
partner_records = request.env[
str(model_name)
].search_read(
domain=[],
fields=fields
) )
# Manually convert datetime fields to string format return Response(
for record in partner_records: json.dumps({"records": self.sanitize_records(partner_records)})
for key, value in record.items(): )
if isinstance(value, datetime): if method == "POST":
record[key] = value.isoformat()
data = json.dumps({
'records': partner_records
})
datas.append(data)
return request.make_response(data=datas)
except:
return ("<html><body><h2>Invalid JSON Data"
"</h2></body></html>")
if method == 'POST':
if not option.is_post: if not option.is_post:
return ("<html><body><h2>Method Not Allowed" raise NotImplementedError()
"</h2></body></html>") if not data or "values" not in data:
else: raise ValueError("No Data Provided")
try:
data = json.loads(request.httprequest.data) data = json.loads(request.httprequest.data)
datas = [] new_resource = request.env[str(model_name)].create(data["values"])
new_resource = request.env[str(model_name)].create( partner_records = request.env[str(model_name)].search_read(
data['values']) [("id", "=", new_resource.id)], fields
partner_records = request.env[ )
str(model_name)].search_read( return Response(
domain=[('id', '=', new_resource.id)], json.dumps({"new_record": self.sanitize_records(partner_records)}),
fields=fields status=201,
) )
new_data = json.dumps({'New resource': partner_records, }) if method == "PUT":
datas.append(new_data)
return request.make_response(data=datas)
except:
return ("<html><body><h2>Invalid JSON Data"
"</h2></body></html>")
if method == 'PUT':
if not option.is_put: if not option.is_put:
return ("<html><body><h2>Method Not Allowed" raise NotImplementedError()
"</h2></body></html>")
else: if "id" not in query:
if rec_id == 0: raise ValueError("No ID Provided")
return ("<html><body><h2>No ID Provided" if not data or "values" not in data:
"</h2></body></html>") raise ValueError("No Data Provided")
else:
resource = request.env[str(model_name)].browse( resource_id = str(query.get("id"))
int(rec_id)) resource = request.env[str(model_name)].browse(int(resource_id))
if not resource.exists(): if not resource.exists():
return ("<html><body><h2>Resource not found" raise ValueError("Resource not found")
"</h2></body></html>")
else:
try:
datas = []
data = json.loads(request.httprequest.data) data = json.loads(request.httprequest.data)
resource.write(data['values']) resource.write(data["values"])
partner_records = request.env[ partner_records = request.env[str(model_name)].search_read(
str(model_name)].search_read( [("id", "=", resource.id)], fields
domain=[('id', '=', resource.id)], )
fields=fields return Response(
) json.dumps(
new_data = json.dumps( {"updated_record": self.sanitize_records(partner_records)}
{'Updated resource': partner_records, )
}) )
datas.append(new_data) if method == "DELETE":
return request.make_response(data=datas)
except:
return ("<html><body><h2>Invalid JSON Data "
"!</h2></body></html>")
if method == 'DELETE':
if not option.is_delete: if not option.is_delete:
return ("<html><body><h2>Method Not Allowed" raise NotImplementedError()
"</h2></body></html>")
else: if "id" not in query:
if rec_id == 0: raise ValueError("No ID Provided")
return ("<html><body><h2>No ID Provided"
"</h2></body></html>") resource_id = str(query.get("id"))
else: resource = request.env[str(model_name)].browse(int(resource_id))
resource = request.env[str(model_name)].browse(
int(rec_id))
if not resource.exists(): if not resource.exists():
return ("<html><body><h2>Resource not found" raise ValueError("Resource not found")
"</h2></body></html>")
else:
records = request.env[ partner_records = request.env[str(model_name)].search_read(
str(model_name)].search_read( [("id", "=", resource.id)], fields
domain=[('id', '=', resource.id)],
fields=['id', 'display_name']
) )
remove = json.dumps(
{"Resource deleted": records,
})
resource.unlink() resource.unlink()
return request.make_response(data=remove) return Response(
json.dumps(
{
"message": "Resource deleted",
"data": self.sanitize_records(partner_records),
}
),
status=202,
)
# If not using any method above, simply return an error
raise NotImplementedError()
except ValueError as e:
return Response(json.dumps({"message": e.args[0]}), status=403)
except NotImplementedError as e:
return Response(
json.dumps(
{
"message": f"Method not allowed. {e.args[0]}Please contact your admininstrator to enable {method} method for {model_display_name or 'this'} record."
}
),
status=405,
)
except Exception:
return Response(
json.dumps({"message": "Internal server error"}), status=500
)
@http.route(['/send_request'], type='http', @http.route(
auth='none', ["/send_request"],
methods=['GET', 'POST', 'PUT', 'DELETE'], csrf=False) type="http",
auth="none",
methods=["GET", "POST", "PUT", "DELETE"],
csrf=False,
)
def fetch_data(self, **kw): def fetch_data(self, **kw):
"""This controller will be called when sending a request to the """This controller will be called when sending a request to the
specified url, and it will authenticate the api-key and then will specified url, and it will authenticate the api-key and then will
generate the result""" generate the result"""
http_method = request.httprequest.method http_method = request.httprequest.method
api_key = request.httprequest.headers.get('api-key') api_key = request.httprequest.headers.get("api-key")
auth_api = self.auth_api_key(api_key) auth_api = self.auth_api_key(api_key)
model = kw.get('model') model = kw.pop("model")
username = request.httprequest.headers.get('login') username = request.httprequest.headers.get("login")
password = request.httprequest.headers.get('password') password = request.httprequest.headers.get("password")
request.session.authenticate(request.session.db, username, request.session.authenticate(request.session.db, username, password)
password) model_id = request.env["ir.model"].search([("model", "=", model)])
model_id = request.env['ir.model'].search(
[('model', '=', model)])
if not model_id: if not model_id:
return ("<html><body><h3>Invalid model, check spelling or maybe " return Response(
"the related " json.dumps(
"module is not installed" {
"</h3></body></html>") "message": "Invalid model, check spelling or maybe the related module is not installed"
}
if auth_api == True: ),
if not kw.get('Id'): status=403,
rec_id = 0 )
else:
rec_id = int(kw.get('Id')) if auth_api.status_code == 200:
result = self.generate_response(http_method, model_id.id, rec_id) result = self.generate_response(http_method, model=model_id.id, **kw)
return result return result
else: else:
return auth_api return auth_api
@http.route(['/odoo_connect'], type="http", auth="none", csrf=False, @http.route(
methods=['GET']) ["/odoo_connect"], type="http", auth="none", csrf=False, methods=["GET"]
)
def odoo_connect(self, **kw): def odoo_connect(self, **kw):
"""This is the controller which initializes the api transaction by """This is the controller which initializes the api transaction by
generating the api-key for specific user and database""" generating the api-key for specific user and database"""
username = request.httprequest.headers.get('login') username = request.httprequest.headers.get("login")
password = request.httprequest.headers.get('password') password = request.httprequest.headers.get("password")
db = request.httprequest.headers.get('db') db = request.httprequest.headers.get("db")
try: try:
request.session.update(http.get_default_session(), db=db) request.session.update(http.get_default_session(), db=db)
auth = request.session.authenticate(request.session.db, username, auth = request.session.authenticate(request.session.db, username, password)
password) user = request.env["res.users"].browse(auth)
user = request.env['res.users'].browse(auth)
api_key = request.env.user.generate_api(username) api_key = request.env.user.generate_api(username)
datas = json.dumps({"Status": "auth successful", datas = json.dumps(
"User": user.name, {"Status": "auth successful", "User": user.name, "api-key": api_key}
"api-key": api_key}) )
return request.make_response(data=datas) return Response(datas)
except: except Exception:
return ("<html><body><h2>wrong login credentials" return Response(
"</h2></body></html>") json.dumps({"message": "wrong login credentials"}), status=401
)

Write Preview
Loading…
Cancel
Save