sitetracker-python 0.1.0

A simple wrapper to make the SiteTracker API easier to use.

SiteTracker Wrapper

English

Introduction

Client Python permettant de faciliter / fiabiliser l'usage de l'API REST de SiteTracker pour créer, consulter et modifier des objets. C'est une version simplifiée de SimpleSalesforce adaptée à certains objets de SiteTracker.

Il pourra être utilisé dans le backend d'une application web pour fiabiliser les requêtes vers SiteTracker, ou pour automatiser des tâches en lien avec du traitement de données.

L'avantage de ce wrapper est qu'il ne nécessite aucune dépendance externe hormis requests, pandas et python-dotenv, et qu'il est facilement extensible.

L'objectif est qu'il nous soit utile pour réaliser plusieurs projets différents en lien avec l'API SiteTracker, pour tout ce qui nécessitera de l'extraction / insertion / analyse de données.

Quelques exemples

from sitetracker_python import SiteTracker

st = SiteTracker()

# Faire une requête SOQL
q = st.query("SELECT Id, Name FROM Account LIMIT 5")

# Obtenir les champs de l'objet Contact
fields = st.get_fields("Contact")

# Obtenir uniquement les champs "name" et "label" pour tous les objets
objects = st.get_objects(fields=["name", "label"])

# Obtenir toutes les informations sur l'objet "Account"
account = st.get_object_by_name("Account")

# Créer un nouveau contact
data = {
    "FirstName": "Rayan",
    "LastName": "TEST",
    "Phone": "[phone]",
}

success, object_id = st.create_record("Contact", data)

# Mettre à jour un enregistrement
new_data = {
   "LastName": "new_name",
}

success, response = st.update_record("Contact", object_id, new_data) # id de l'objet précédemment créé.

Quickstart

Installation

1. Créer un environnement virtuel et l'activer

python -m venv .venv
.venv/Scripts/activate

(.venv) devrait apparaître dans votre terminal, si c'est le cas, l'environnement virtuel est activé.

TODO installation via pip

Si tout s'est bien passé, vous pouvez maintenant importer la librairie dans votre code.

Variables d'environnement

Pour s'authentifier auprès de l'API, il est nécessaire de fournir 4 variables:

SANDBOX_USER_EMAIL=EMAIL.TEST_DOMAIN
SANDBOX_USER_PASSWORD=PASSWORD
SANDBOX_SECURITY_TOKEN=TOKEN

PRODUCTION_USER_EMAIL=EMAIL
PRODUCTION_USER_PASSWORD=PASSWORD
PRODUCTION_SECURITY_TOKEN=TOKEN

Un fichier .env.example est fourni. Il suffit de modifier les valeurs et de renommer le fichier en .env.

Les variables SANDBOX_SECURITY_TOKEN et PRODUCTION_SECURITY_TOKEN sont relatives à votre propre compte. Le jeton peut être créé / obtenu via les paramètres de votre compte SiteTracker > Réinitialiser mon jeton de sécurité.

⚠️ Important

Salesforce détermine automatiquement l'environnement de test ou de production. Cela est fait en fonction des informations d'identification utilisées pour se connecter.

Si je fournis des identifiants relatifs à un environnement de staging, il suffit simplement de définir sandbox=True lors de l'initialisation du wrapper et de fournir dans le .env les variables d'environnement correspondant à cet environnement.

Si je veux me connecter à l'environnement de production, les noms des variables d'environnement ne changent pas, mais les valeurs doivent être celles relatives à la production.

Utilisation

Lors de l'initialisation d'une instance, la connexion se fait automatiquement. L'authentification se fait via le protocole SOAP. Une fois que celle-ci est réalisée, un objet de type requests.Session est créé et utilisé pour effectuer les requêtes.

from sitetracker_python import SiteTracker

st = SiteTracker() # prêt à effectuer des requêtes

Par défaut, l'instance est configurée pour se connecter à l'environnement de test (sandbox). Pour se connecter à l'environnement de production, il suffit de passer sandbox=False lors de l'initialisation.

st = SiteTracker(sandbox=False)

Il est également possible de spécifier la version de l'API que l'on souhaite utiliser, sous forme de string. Par défaut, la version utilisée est la 60.0, proche de la dernière. Notez que l'on ne mentionne pas le v dans la version. La liste des versions est disponible ici.

st = SiteTracker(st_version="50.0")

Extension par héritage

Vous pouvez étendre la classe SiteTracker pour ajouter vos méthodes métier sans modifier le cœur de la librairie.

from sitetracker_python import SiteTracker


class CustomSiteTracker(SiteTracker):
    def get_account_names(self, limit: int = 10) -> list[str]:
        result = self.query(f"SELECT Name FROM Account LIMIT {limit}")
        if isinstance(result, list):  # payload d'erreur Salesforce
            return []
        return [
            r.get("Name")
            for r in result.get("records", [])
            if isinstance(r, dict) and r.get("Name")
        ]

    def contact_exists_by_email(self, email: str) -> bool:
        contact_id = self.check_instance_exists("Contact", "Email", email)
        return contact_id is not None

Exemple complet: voir le fichier examples.py.

Endpoints et exemples

Les endpoints sont séparés en 3 catégories: GET, POST et UPDATE. Les endpoints GET sont utilisés pour obtenir des informations sur les objets, les champs, les requêtes SOQL, etc. Les endpoints POST sont utilisés pour créer, modifier ou supprimer des objets. Les endpoints UPDATE sont utilisés pour mettre à jour des objets existants.

Sommaire

• GET

  • query
  • fetch_endpoint
  • get_fields
  • get_objects
  • get_object_by_name
  • get_object_records
  • get_picklist_values
  • get_all_fields_values
  • get_record_by_id
  • get_specific_object_relation
  • check_instance_exists
  • find_duplicates
  • del_condition_valeur

• POST

  • create_record

• UPDATE

  • update_record

• DELETE

  • delete_record

• ATTACHMENTS

  • upload_file_no_lease
  • get_attachments_of_activities
  • delete_attachment_by_id

GET

query(soql: str, include_deleted: bool = False) -> dict | list

Effectue une requête SOQL. Si include_deleted=True, la méthode utilise l'endpoint queryAll.

q = st.query("SELECT Id, Name FROM Account LIMIT 2")

Exemple de réponse

{
   "done": true,
   "records": [
      {
         "Id": "001Aa00000cPSjEIAW",
         "Name": "SNEF",
         "attributes": {
            "type": "Account",
            "url": "/services/data/v60.0/sobjects/Account/001Aa00000cPSjEIAW"
         }
      },
      {
         "Id": "001Aa00000cPSjFIAW",
         "Name": "ENEDIS",
         "attributes": {
            "type": "Account",
            "url": "/services/data/v60.0/sobjects/Account/001Aa00000cPSjFIAW"
         }
      }
   ],
   "totalSize": 2
}

fetch_endpoint(endpoint: str) -> dict | list

Permet d'interroger directement un endpoint Salesforce relatif à l'API REST.

describe = st.fetch_endpoint("sobjects/Account/describe")

Exemple d'erreur sur endpoint invalide:

error_payload = st.fetch_endpoint("sobjects/Account/describe/BAD_URL")

La méthode retourne le JSON brut renvoyé par l'API (souvent un dict, parfois une list d'erreurs).

get_fields(object_name: str) -> list

Retourne l'ensemble des champs d'un objet donné. Si l'objet n'existe pas, une erreur de type ObjectDoesNotExist est levée.

fields = st.get_fields("Contact")

Exemple de réponse:

{
   "label": "ID du compte",
   "name": "AccountId",
   "referenceTo": [
      "Account"
   ],
   "type": "reference"
}

get_objects(fields: list[str], limit: int) -> list

Récupère les objets SiteTracker disponibles pour l'utilisateur connecté. La structure d'un objet peut être consultée dans data/all_objects.json. Si on souhaite uniquement obtenir certains champs (fields), il suffit de les spécifier dans la liste fields. Il est également possible de limiter le nombre d'objets retournés en spécifiant limit.

Si limit n'est pas spécifié, l'ensemble des objets retourné par l'API est renvoyé.

# Tous les objets avec tous les champs
objects = st.get_objects()

# Tous les objets avec uniquement les champs "name" et "label".
objects_with_name_and_label = st.get_objects(fields=["name", "label"])

# Les 5 premiers objets avec uniquement le champ "name".
only_5_objects_and_their_name = st.get_objects(fields=["name"], limit=5)

get_object_by_name(object_name: str) -> dict

Récupère toutes les informations sur un objet donné. Si l'objet n'existe pas, une erreur de type ObjectDoesNotExist est levée.

account = st.get_object_by_name("Account")

get_object_records(object_name: str, fields: list[str], limit: int) -> dict

Récupère les enregistrements d'un objet Salesforce spécifique. Il est possible de spécifier les champs à retourner en utilisant la liste fields. Si fields n'est pas spécifié, tous les champs sont retournés.

On peut également limiter le nombre d'enregistrements retournés en spécifiant limit.

# Obtenir la liste des projets et leur nom en mitant à 1
projects = st.get_object_records("sitetracker__Project__c", ["Name"], limit=1)

Exemple de réponse:

{
   "totalSize":1,
   "done":true,
   "records":[
      {
         "attributes":{
            "type":"sitetracker__Project__c",
            "url":"/services/data/v60.0/sobjects/sitetracker__Project__c/a0oAa0000044wuLIAQ"
         },
         "Name":"P-000850"
      }
   ]
}

get_picklist_values(object_api_name: str, field_api_name: str) -> dict

Retourne les valeurs d'une picklist pour un champ donné.

values = st.get_picklist_values("Account", "Type")
print(values)

Format de retour:

{
  "Customer": { "label": "Customer", "is_active": true },
  "Prospect": { "label": "Prospect", "is_active": true }
}

Si le champ n'existe pas ou n'est pas une picklist, la méthode retourne {}.

get_all_fields_values(object_name: str, fields: list[str], where: Optional[str] = None, include_deleted: bool = False, as_dataframe: bool = False) -> list[dict] | pd.DataFrame

Récupère toutes les lignes d'un objet pour une liste de champs, avec pagination automatique.

• Id est toujours inclus dans le résultat.
• where permet d'ajouter une clause SOQL (sans le mot-clé WHERE).
• include_deleted=True utilise queryAll.
• as_dataframe=True retourne un pandas.DataFrame.

rows = st.get_all_fields_values(
    object_name="Account",
    fields=["Name", "Type"],
    where="IsDeleted = false",
)

df = st.get_all_fields_values(
    object_name="Account",
    fields=["Name", "Type"],
    as_dataframe=True,
)

get_record_by_id(object_name: str, record_id: str) -> Tuple[bool, dict]

Récupère un enregistrement d'un objet Salesforce spécifique en utilisant son ID. Retourne un booléen correspondant au statut de l'opération ainsi que l'enregistrement ou un message d'erreur.

success, record = st.get_record_by_id("sitetracker__Project__c", "a0oAa0000044wuLIAQ")

if success is True:
   print(record)
elif success is False:
   print(record) # Une erreur non prévue s'est produite ...

get_specific_object_relation(object_name: str, record_id: str, related_field: str) -> dict | Tuple[bool, dict]:

Récupère les informations d'un objet lié à un objet spécifique. Par exemple, si on souhaite obtenir les informations d'un site lié à un projet, on peut utiliser cette fonction.

# Obtenir les informations du Site du Projet avec l'ID "a0oG5000000IvJBIA0"
success, relation = st.get_specific_object_relation("sitetracker__Project__c", "a0oG5000000IvJBIA0", "sitetracker__Site__c")

if success is True:
    print(relation)
else:
    # En cas d'erreur, afficher le message exact
    print(relation)

check_instance_exists(object_name: str, field_name: str, field_value: str) -> Optional[str]

Vérifie l'existence d'un enregistrement à partir d'un champ donné, et retourne l'Id trouvé (ou None).

account_id = st.check_instance_exists("Account", "Name", "ACME")
if account_id:
    print("Existe déjà:", account_id)

find_duplicates(object_name: str, field_value: str) -> dict[Any, int]

Recherche les doublons sur un champ (valeurs non nulles) et retourne un dictionnaire {valeur: nombre_occurrences}.

duplicates = st.find_duplicates("Account", "Name")
print(duplicates)

del_condition_valeur(object_name: str, field_value: str, list_valeurs: list[str]) -> int

Supprime les enregistrements dont le champ field_value est dans list_valeurs. Retourne le nombre d'enregistrements supprimés.

deleted_count = st.del_condition_valeur(
    object_name="Account",
    field_value="External_Id__c",
    list_valeurs=["EXT-001", "EXT-002"],
)
print(deleted_count)

POST

create_record(object_name: str, data: dict) -> Tuple[bool, str]

Crée un nouvel enregistrement d'un objet SiteTracker spécifique. Pour créer un enregistrement, il est nécessaire de fournir le nom de l'objet ainsi que l'ensemble des champs et valeurs à insérer. Si l'objet n'existe pas, une erreur de type ObjectDoesNotExist est levée.

object_name (str): Le nom de l'objet SiteTracker dans lequel vous souhaitez créer un enregistrement. data (dict): Un dictionnaire contenant les données à insérer dans SiteTracker.

La fonction retourne un tuple de deux éléments. Le premier élément est un booléen indiquant si la création de l'enregistrement a réussi ou non. Le second élément est soit l'ID de l'enregistrement créé, soit un message d'erreur renvoyé par SiteTracker.

data = {
    "FirstName": "Rayan",
    "LastName": "TEST",
    "Phone": "[phone]",
}
success, new_object = st.create_record("Contact", data)

if success is True:
    print(f"Record created with id: {new_object}")
else:
    print("Record creation failed")
    print(new_object)

Exemple de création de plusieurs enregistrements avec un suivi des succès et des erreurs.

data = [...]

new_records = []
errors = []

for record in data:
    success, id_ = st.create_record("Contact", record)
    if success is True:
        new_records.append(id_)
    else:
        errors.append({record["LastName"]: id_}) # En partant du principe que "LastName" fait partie des champs du contact

print(f"{len(new_records)} records created with success")
print(new_records)

print(f"{len(errors)} records failed")
print(errors)

UPDATE

update_record(object_name: str, record_id: str, data: dict) -> Tuple[bool, str]

data_to_insert = {
      "Lien_PV_Design__c": nouveau_lien,
      "Lien_XRP__c": nouveau_lien,
      "Lien_Sharepoint__c": nouveau_lien,
      "Lien_Hubspot__c": nouveau_lien,
   }

success, response = st.update_record(
   "sitetracker__Project__c", id_, data_to_insert
)

print(success, response) # True, "Mise à jour effectuée avec succès."

Si la modification de l'objet se passe comme prévu, la fonction retournera True, "Mise à jour effectuée avec succès.". Sinon, elle retournera False ainsi que le message d'erreur retourné par SiteTracker.

Notez que les noms des champs dans la variable data doivent correspondre aux noms des champs de l'objet SiteTracker.

DELETE

delete_record(object_name: str, record_id: str) -> Tuple[bool, str]

Supprime un enregistrement d'un objet SiteTracker spécifique en utilisant son ID.

success, response = st.delete_record("sitetracker__Project__c", "a0oAa0000044wuLIAQ")

if success is True:
    print("Record deleted successfully")
else:
    print("Record deletion failed")
    print(response)

ATTACHMENTS

upload_file_no_lease(activity_sf_id: str, b64_data: str, filename: str, file_extension: str | None = None) -> dict

Upload un fichier lié à une activité Sitetracker (hors objets de type Lease), en créant:

• ContentVersion
• ContentDocument
• sitetracker__Attachment__c

result = st.upload_file_no_lease(
    activity_sf_id="a0oXXXXXXXXXXXXXXX",
    b64_data=base64_payload,
    filename="rapport",
    file_extension="pdf",
)
print(result["success"])

Le résultat contient notamment:

• success
• sitetracker_attachment_id
• content_version_id
• content_document_id
• filename
• error (en cas d'échec)

get_attachments_of_activities(activity_ids: list[ActivityIDInstance], chunk_size: int = 150) -> dict[str, list[dict]]

Récupère les pièces jointes pour une liste d'activités (requêtes en lot).

from sitetracker_python.models import ActivityIDInstance

activity_rows = [
    ActivityIDInstance(
        activity_id="a1HXXXXXXXXXXXXXXX",
        sitetracker_instance_id="P-000001",
    ),
]

attachments = st.get_attachments_of_activities(activity_rows)

Retour: dictionnaire indexé par activity_id, avec une liste d'objets contenant par exemple:

• id
• activity_id
• instance_id
• instance_name
• filename
• author_name
• created_date
• last_modified_date

delete_attachment_by_id(attachment_id: str) -> bool

Supprime complètement une pièce jointe Sitetracker:

• suppression de sitetracker__Attachment__c
• suppression du ContentDocument associé

deleted = st.delete_attachment_by_id("a1JXXXXXXXXXXXXXXX")
print(deleted)  # True si la suppression est complète