Metadata-Version: 2.3
Name: sitetracker-python
Version: 0.1.0
Summary: A simple wrapper to make the SiteTracker API easier to use.
Author: Rayan
Author-email: Rayan <rayan@email.com>
Requires-Dist: pandas>=3.0.1
Requires-Dist: python-dotenv>=1.2.1
Requires-Dist: requests>=2.32.5
Requires-Python: >=3.13
Description-Content-Type: text/markdown

# SiteTracker Wrapper

[English](README.en.md)

## 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](https://github.com/simple-salesforce/simple-salesforce) 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

```python
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": "+33 685448548",
}

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

```bash
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:

```bash
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.

```python
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.

```python
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](https://YOUR_SALESFORCE_INSTANCE.sandbox.my.salesforce.com/services/data/).

```python
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.

```python
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`](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](#get)
  - [query](#queryquery-str---dict)
  - [fetch_endpoint](#fetch_endpointendpoint-str---dict--list)
  - [get_fields](#get_fieldsobject_name-str---list)
  - [get_objects](#get_objectsfields-liststr-limit-int---list)
  - [get_object_by_name](#get_object_by_nameobject_name-str---dict)
  - [get_object_records](#get_object_recordsobject_name-str-fields-liststr-limit-int---dict)
  - [get_picklist_values](#get_picklist_valuesobject_api_name-str-field_api_name-str---dict)
  - [get_all_fields_values](#get_all_fields_valuesobject_name-str-fields-liststr-where-optionalstr--none-include_deleted-bool--false-as_dataframe-bool--false---listdict--pddataframe)
  - [get_record_by_id](#get_record_by_idobject_name-str-record_id-str---tuplebool-dict)
  - [get_specific_object_relation](#get_specific_object_relationobject_name-str-record_id-str-related_field-str---dict--tuplebool-dict)
  - [check_instance_exists](#check_instance_existsobject_name-str-field_name-str-field_value-str---optionalstr)
  - [find_duplicates](#find_duplicatesobject_name-str-field_value-str---dictany-int)
  - [del_condition_valeur](#del_condition_valeurobject_name-str-field_value-str-list_valeurs-liststr---int)

- [POST](#post)
   - [create_record](#create_recordobject_name-str-data-dict---tuplebool-str)

- [UPDATE](#update)
   - [update_record](#update_recordobject_name-str-record_id-str-data-dict---tuplebool-str)

- [DELETE](#delete)
   - [delete_record](#delete_recordobject_name-str-record_id-str---tuplebool-str)

- [ATTACHMENTS](#attachments)
   - [upload_file_no_lease](#upload_file_no_leaseactivity_sf_id-str-b64_data-str-filename-str-file_extension-str--none--none---dict)
   - [get_attachments_of_activities](#get_attachments_of_activitiesactivity_ids-listactivityidinstance-chunk_size-int--150---dictstr-listdict)
   - [delete_attachment_by_id](#delete_attachment_by_idattachment_id-str---bool)

## GET

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

Effectue une requête [SOQL](https://developer.salesforce.com/docs/atlas.en-us.soql_sosl.meta/soql_sosl/storce_api_calls_soql.htm).
Si `include_deleted=True`, la méthode utilise l'endpoint `queryAll`.

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

Exemple de réponse

```json
{
   "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.

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

Exemple d'erreur sur endpoint invalide:

```python
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.

```python
fields = st.get_fields("Contact")
```

Exemple de réponse:

```json
{
   "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é.

```python
# 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.

```python
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`.

```python
# 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:

```json
{
   "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é.

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

Format de retour:

```json
{
  "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`.

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

```python
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.

```python
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.

```python
# 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`).

```python
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}`.

```python
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.

```python
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.

```python
data = {
    "FirstName": "Rayan",
    "LastName": "TEST",
    "Phone": "+33 685448548",
}
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.

```python
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]

```python
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.

```python
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`

```python
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).

```python
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é

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