cas/civicrmapi4
0
0
Fork 1
Code Issues Pull Requests Projects Releases Wiki Activity

Reorg and major documentation update

Browse Source
This commit is contained in:
Cassowary Rusnov 2021-11-22 10:26:53 -08:00
parent 616f220cfc
commit d35a35f6ad
3 changed files with 260 additions and 132 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
src/civicrmapi4/__init__.py → civicrmapi4/__init__.py
View File

260
civicrmapi4/civicrmapi4.py Normal file
View File

@ -0,0 +1,260 @@
"""Main module."""
import json
import requests
import logging
from typing import List, Optional, Dict, Union
from bs4 import BeautifulSoup
log = logging.getLogger(__name__)
# TODOs and FIXMEs
#
# * come up with a better way to specify `where` clauses.
#
class APIError (BaseException):
"""
Thrown for generic API errors such as at login and other similar circumstances.
This is also the base class for other errors.
"""
...
class CallFailed(APIError):
"""Thrown when an API call fails, specifically."""
...
class APIv4:
"""
Wrapper for all APIv4 calls. Created with a base URL string, .login must be called prior
to any API calls being performed.
This uses login session to authenticate with the API (and does not currently support other
methods of authentication), thus a username and password are required.
"""
def __init__(self, base_url: str):
"""
Create a new APIv4 object. Takes a base URL as its one argument, for example
`https://crm.mysite.com`.
Parameters:
base_url (str): The base URL to the CiviCRM instance.
Raises:
None
"""
self.base_url = base_url
self.api_url = self.base_url + "/civicrm/ajax/api4"
self.session = None
def login(self, user: str, password: str) -> None:
"""
Perform a login operation against target CRM. Stores session internally so generally
only one login is required per session.
FIXME should raise on login failure or other things being amiss.
Parameters:
user (str): The user name to use to authenticate with the CRM.
password (str): The password to use to authenticate with the CRM.
Returns:
None
Raises:
None
"""
self.session = requests.Session()
### Perform login post
self.session.cookies.update({"has_js": "1"})
postdata = {
"name": user,
"pass": password,
"form_id": "user_login_block",
}
res = self.session.post(self.base_url, data=postdata)
if (not res.ok):
log.error("Login failed. Code: {}. Result: {}".format(res.status_code, res.text))
# we should raise an error here.
self.session = None
return
# session now has the correct session cookie to do authenticated requests
# these headers seem to be required to pass CSRF
self.session.headers.update({
"X-Requested-With": "XMLHttpRequest",
"Content-Type": "application/json;charset=utf-8",
})
log.info("Successfully logged into CRM.")
def _call(self, objName: str, action: str, params: Optional[Dict] = None) -> requests.Response:
"""
Low level, minimal processing call to the API.
Parameters:
objName (str): The object name to operate on, for example `Contact`.
action (str): The action to call on the object, for example `get`.
params (dict): A dictionary of parameters to pass to the API.
Returns:
:class:`requests.Response`: A response object from requests.
Raises:
None
"""
if (self.session is None):
log.error("Need to login first.")
raise APIError("Need to login first.")
url = self.api_url+'/'+objName+'/'+action
if params:
result = self.session.post(url, params={'params':json.dumps(params)}, data="")
else:
result = self.session.post(url, data="")
if (not result.ok):
raise CallFailed("Call returned non-2xx: {} {}".format(result.status_code, result.text))
return result.json()
def call_values(self, objName: str, action: str, params: Optional[Dict] = None) -> Optional[List]:
"""
A wrapper for `call` which returns the most common parts of the response for most uses.
Parameters:
objName (str): The object name to operate on, for example `Contact`.
action (str): The action to call on the object, for example `get`.
params (dict): A dictionary of parameters to pass to the API.
Returns:
`list` or `None`: The values returned from the API call (or None if not present).
"""
result = self._call(objName, action, params)
if result and ("values" in result):
return result["values"]
return None
def get_fields(self, objName: str) -> List:
"""
Return a list of fields on the target object type.
Parameters:
objName (str): The object name to operate on, for example `Contact`.
Returns:
`list` or `None`: The list of fields if successful, or None.
"""
return self.call_values(objName, "getFields")
def get_actions(self, objName: str) -> List:
"""
Return a list of possible actions for a given object type.
Parameters:
objName (str): The object name to operate on, for example `Contact`.
Returns:
`list` or `None`: The list of possible actions, or None.
"""
return self.call_values(objName, "getActions")
def get_entities(self, detailed: bool = False) -> Union[List[str], List[Dict]]:
"""
Return a list of possible entities.
Parameters:
detailed (bool, default `False`): Wether to include full metadata about entities.
Returns:
`list` or `None`: The list of entities (objects) that can be operated on.
"""
ent = self.call_values("Entity", "get")
if detailed:
return ent
return [x["name"] for x in ent]
def get(self, objName: str, where: Optional[List]=None, limit: Optional[int]=None, select=Optional[List[str]]) -> List:
"""
Get objects based on a where clause, optional limit and optional field list.
Parameters:
objName (str): The object name to operate on, for example `Contact`.
where (optional, list): A list of lists, each list specifies a `where` clause such as ["name", "=", "Foo"]
limit (optional, int): A number specifying the maximum number of rows to return.
select (optional, list): A list of field names to return rather than all fields.
Returns:
`list` of `dict` or None: A list of object dictionries matching the above parameters.
"""
params = {}
if limit is not None:
params["limit"] = limit
if where is not None:
params["where"] = where
if columns is not None and columns:
params["select"] = select
return self.call_values(objName, "get", params=params)
def create(self, objName: str, values: Dict) -> Dict:
"""
Create an object using the values specified, and return the newly created object.
Parameters:
objName (str): The object name to operate on, for example `Contact`.
values (dict): A dictionary specifying a subset of the object's fields and the values.
Returns:
`dict`: The full values of the newly-created object.
"""
result = self.call_values(objName, "create", params={"values": values})
if result:
return result[0]
return {}
def delete(self, objName: str, where: List) -> int:
"""
Delete objects with the specified where clauses.
Parameters:
objName (str): The object name to operate on, for example `Contact`.
where (optional, list): A list of lists, each list specifies a `where` clause such as ["name", "=", "Foo"]
Returns:
`int`: The count of the number of objects deleted.
"""
result = self._call(objName, "delete", params={"where": where})
if "count" in result:
return result["count"]
return 0
def replace(self, objName: str, records: List[Dict], where: List) -> List:
# FIXME
# (upsert operation)
...
def update(self, objName: str, fields: Dict, where: List) -> List:
"""Update is an upsert operation."""
# FIXME
# (update operation)
...

132
src/civicrmapi4/civicrmapi4.py
View File

@ -1,132 +0,0 @@
"""Main module."""
import json
import requests
import logging
from typing import List, Optional, Dict, Union
from bs4 import BeautifulSoup
log = logging.getLogger(__name__)
class APIError (BaseException):
...
class CallFailed(APIError):
...
class APIv4:
def __init__(self, base_url: str):
self.base_url = base_url
self.api_url = self.base_url + "/civicrm/ajax/api4"
self.session = None
def login(self, user: str, password: str):
self.session = requests.Session()
### Perform login post
self.session.cookies.update({"has_js": "1"})
postdata = {
"name": user,
"pass": password,
"form_id": "user_login_block",
}
res = self.session.post(self.base_url, data=postdata)
if (not res.ok):
log.error("Login failed. Code: {}. Result: {}".format(res.status_code, res.text))
# we should raise an error here.
self.session = None
return
# session now has the correct session cookie to do authenticated requests
# these headers seem to be required to pass CSRF
self.session.headers.update({
"X-Requested-With": "XMLHttpRequest",
"Content-Type": "application/json;charset=utf-8",
})
log.info("Successfully logged into CRM.")
def _call(self, objName: str, action: str, params: Optional[Dict] = None) -> requests.Response:
"""
Low level, minimal processing call to the API.
"""
if (self.session is None):
log.error("Need to login first.")
raise APIError("Need to login first.")
url = self.api_url+'/'+objName+'/'+action
if params:
result = self.session.post(url, params={'params':json.dumps(params)}, data="")
else:
result = self.session.post(url, data="")
if (not result.ok):
raise CallFailed("Call returned non-2xx: {} {}".format(result.status_code, result.text))
return result.json()
def call_values(self, objName: str, action: str, params: Optional[Dict] = None) -> Optional[List]:
"""A quick wrapper to just return the values of an API call"""
result = self._call(objName, action, params)
if result and ("values" in result):
return result["values"]
return None
def get_fields(self, objName: str) -> List:
"""
Return a list of fields on the target object type.
"""
return self.call_values(objName, "getFields")
def get_actions(self, objName: str) -> List:
"""
Return a list of possible actions for a given object type.
"""
return self.call_values(objName, "getActions")
def get_entities(self, detailed: bool = False) -> Union[List[str], List[Dict]]:
"""Return a list of possible entities."""
ent = self.call_values("Entity", "get")
if detailed:
return ent
return [x["name"] for x in ent]
def get(self, objName: str, where: Optional[List]=None, limit: Optional[int]=None, select=Optional[List[str]]) -> List:
"""Get objects based on a where clause, optional limit and optional field list."""
params = {}
if limit is not None:
params["limit"] = limit
if where is not None:
params["where"] = where
if columns is not None and columns:
params["select"] = select
return self.call_values(objName, "get", params=params)
def create(self, objName: str, values: Dict) -> Dict:
"""Create an object using the values specified, and return the newly created object."""
result = self.call_values(objName, "create", params={"values": values})
if result:
return result[0]
return {}
def delete(self, objName: str, where: List) -> int:
"""Delete objects with the specified where clauses and return the count.."""
result = self._call(objName, "delete", params={"where": where})
if "count" in result:
return result["count"]
return 0
def replace(self, objName: str, records: List[Dict], where: List) -> List:
...
def update(self, objName: str, fields: Dict, where: List) -> List:
"""Update is an upsert operation."""
...