Updating Data Sources with Python
If you wish to update data sources inside Element using Element451’s API, here are some sample Python scripts to assist you in this task.
NOTE: If you are a managed integration customer, do not manually or automatically touch data sources that Element451 has synced for you. You can cause issues with the managed integration process by doing so.
Building Blocks#
Element451 API
import json
from dataclasses import dataclass, field
class APIException(Exception):
def __init__(self, status_code, content):
self.status_code = status_code
self.content = json.loads(content)
message = f"Status code: {status_code}\nContent:\n{content}"
super().__init__(message)
@dataclass
class ElementAPI:
api_url: str = field(default=None)
client: str = field(default=None)
feature: str = field(default=None)
service: str = field(default=None)
auth: str = field(default=None)
params: dict = field(default_factory=dict)
headers: dict = field(default_factory=dict)
def __post_init__(self):
self.api_url = f"https://{self.client}.{self.api_url}"
if self.service:
self.params["service"] = self.service
if self.feature:
self.headers["feature"] = self.feature
if self.auth:
self.headers["authorization"] = self.auth
def _handle_fail_request(self, response):
try:
response_content = response.json()
if isinstance(response_content, dict):
content = json.dumps(response_content, indent=4)
else:
content = response_content
except (json.decoder.JSONDecodeError, ValueError):
content = response.text
raise APIException(response.status_code, content)
Data Sources API
import json
from dataclasses import dataclass, field
from typing import List
import requests
from .element_api import ElementAPI
@dataclass
class ElementDataSource(ElementAPI):
"""
Handler for Element Data Sources.
- for regular ds provide ds_guid
- for others provide ds entity name
"""
entity: str = ""
route: str = ""
ds_guid: str = ""
data: List[dict] = field(default_factory=dict)
column_names: List[str] = field(default_factory=list)
def __post_init__(self):
super().__post_init__()
if self.ds_guid:
self.route = f"data/sources/{self.ds_guid}"
if self.route or self.entity:
self.fetch_data_source()
def fetch_data_source(self) -> List[dict]:
data = self._get_data_source_items()
self.column_names = get_columns(data)
self._prepare_data(data)
def _get_data_source_items(self) -> List[dict]:
"""Returns list of specified entities"""
if self.route:
url = f"{self.api_url}/v2/{self.route}/items"
else:
url = f"{self.api_url}/v2/{self.entity}"
response = requests.get(
url,
params=self.params,
headers=self.headers,
)
if response.status_code not in [200, 201]:
self._handle_fail_request(response)
return response.json()["data"]
def _prepare_data(self, data: List[dict]) -> List[dict]:
"""For some items API returns list of elements, and in csv files we need those elements in one string"""
for row in data:
for k, v in row.items():
if isinstance(v, list):
row[k] = parse_list_to_string(v)
self.data = data
def upload_csv(self, file_path: str, mode: str = "upsert") -> dict:
"""Uploads csv document to update specified data source"""
if self.route:
url = f"{self.api_url}/v2/{self.route}/csvimport"
else:
url: str = f"{self.api_url}/v2/{self.entity}/csvimport"
payload = {"mode": mode}
filename = file_path.split("/")[-1]
files = [("import", (filename, open(file_path, "rb"), "text/csv"))]
response = requests.post(
url=url, data=payload, headers=self.headers, files=files, params=self.params
)
if response.status_code not in [200, 201]:
self._handle_fail_request(response)
return response.json()
def create_regular_data_source(
self, name: str, dimensions: List[str], value: str, label: str
) -> dict:
"""
Creates a new regular data source
Example:
Args:
name="TestRegularDS",
dimensions=[
{"name": "The Label", "key": "column_1", "type": "string"},
{"name": "The Value", "key": "column_2", "type": "string"},
{"name": "Other", "key": "column_3", "type": "string"},
],
label="column_1",
value="column_2",
"""
url = f"{self.api_url}/v2/data/sources"
data = {
"active": True,
"name": name,
"dimensions": dimensions,
"value_field": value,
"label_field": label,
"sortby_order": "ascending",
"sortby_field": "",
"type": "regular",
"visible_by": "user",
}
payload = {"source": json.dumps(data)}
headers = self.headers.copy()
headers["Content-Type"] = "application/x-www-form-urlencoded"
response = requests.post(url, headers=headers, data=payload)
if response.status_code not in [200, 201]:
self._handle_fail_request(response)
return response.json()
def parse_list_to_string(lst: List[str]) -> str:
"""Returns string of elements, joined by comma, from list of elements"""
return ",".join([str(n) for n in lst])
def get_columns(data: List[dict]) -> List[str]:
"""Returns list of column names from provided list of dicts."""
return list(set(key for item in data for key in item.keys()))
CSV Handlers
import csv
import os
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import List, Union
@dataclass
class CsvHandler(ABC):
filename: str = "datasource"
file_path: str = f"{os.getcwd()}/data_source_extraction/tmp/files"
data: List[dict] = field(default_factory=list)
column_names: List = field(default_factory=list)
def __post_init__(self):
if ".csv" not in self.filename:
self.filename = f"{self.filename}.csv"
if not os.path.isdir(self.file_path):
os.makedirs(self.file_path, exist_ok=True)
self.file_path = f"{self.file_path}/{self.filename}"
@abstractmethod
def join_data(self) -> None:
pass
def create_csv(self) -> bool:
created: bool = False
with open(self.file_path, "w") as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=self.column_names)
writer.writeheader()
writer.writerows(self.data)
created = True
return created
def delete_csv(self) -> bool:
deleted: bool = False
if os.path.exists(self.file_path):
os.remove(self.file_path)
deleted = True
return deleted
def add_column(self, column_names: Union[str, List[str]]) -> None:
if isinstance(column_names, str):
column_names = [column_names]
self.column_names.extend(
column for column in column_names if column not in self.column_names
)
print(f"All column names: {self.column_names}")
def remove_column(self, column_names: Union[str, List[str]]) -> None:
if isinstance(column_names, str):
column_names = [column_names]
self.column_names = [
column for column in self.column_names if column not in self.column_names
]
print(f"All column names: {self.column_names}")
Terms API
import json
from dataclasses import dataclass
import requests
from .element_api import ElementAPI
@dataclass
class ElementTerms(ElementAPI):
def __post_init__(self):
super().__post_init__()
def get_terms(self) -> dict:
url = f"{self.api_url}/v2/terms/"
response = requests.get(url, params=self.params, headers=self.headers)
if response.status_code != 200:
self._handle_fail_request(response)
return response.json()["data"]
def create_term(
self,
active: bool,
name: str,
code: str,
start_date: str,
end_date: str,
academic_year: int,
files=[],
) -> dict:
url = f"{self.api_url}/v2/terms/"
payload = {
"term": json.dumps(
{
"active": active,
"name": name,
"code": code,
"start_date": start_date,
"end_date": end_date,
"academic_year": academic_year,
"files": files,
}
),
}
response = requests.post(url, json=payload, headers=self.headers, files=files)
if response.status_code not in [200, 201]:
self._handle_fail_request(response)
return response.json()["data"]
def update_term(self, term_id: str, **kwargs) -> dict:
url = f"{self.api_url}/v2/terms/{term_id}"
payload = {"term": json.dumps(kwargs)}
response = requests.put(url, json=payload, headers=self.headers)
if response.status_code != 200:
self._handle_fail_request(response)
return response.json()["data"]
Use Cases#
Create/Update Terms
from element.api.term_handlers import ElementTerms
term = {
"active": True,
"name": "Fall 2021",
"code": "F21",
"start_date": "2020-08-24T00:00:00.000+00:00",
"end_date": "2021-12-09T00:00:00.000+00:00",
"academic_year": 2021,
}
client = "tu"
terms_api = ElementTerms(
api_url="api.451.io", client=client, feature="feature-token", auth="auth-token"
)
created_term = terms_api.create_term(**term)
term_id = created_term.get("guid")
updated_term = terms_api.update_term(term_id=term_id, name="FALL 2021", code="FA21")Update/Upsert Terms using Data Source API & CSV Files
from dataclasses import dataclass
from typing import List
from element.api.csv_handlers import CsvHandler
from element.api.data_source_handlers import ElementDataSource
@dataclass
class TermsCsvHandler(CsvHandler):
filename: str = "terms_data_source"
def join_data(self, your_term_data: List[dict], columns: List[str]) -> None:
self.add_column(columns)
self.data.extend(your_term_data)
client = "tu"
ds_api = ElementDataSource(**clients[client], entity="terms")
your_term_data: List[dict] = [
{
"guid": "", # if you want to update specific one you need to provide the guid
"name": "Fall 2021/23",
"code": "F2122",
"start_date": "2020-08-24T00:00:00.000+00:00",
"end_date": "2021-12-09T00:00:00.000+00:00",
"active": False,
}
]
# join the existing data and new data
handler = TermsCsvHandler(data=ds_api.data)
handler.join_data(your_term_data, columns=ds_api.column_names)
handler.create_csv()
upload_info = ds_api.upload_csv(handler.file_path)
print(upload_info)
handler.delete_csv()Was this page helpful?