brahm-centre-utils 1.0.6

Brahm Centre internal utility functions

Brahm Centre Utils

Internal helper package for Brahm Centre engineering teams. It provides ready-to-use utilities for Salesforce, SharePoint, and data-cleaning workflows so that automation projects share one consistent toolkit.

Installation

pip install brahm-centre-utils

# or for local development with tests
pip install -e .[dev]

The package depends on pandas, requests, simple-salesforce, and openpyxl. Installing in a virtual environment is strongly recommended.

Configuration

Salesforce credentials

Store your Salesforce secrets as environment variables or load them from a .env file before passing them into simple_salesforce.Salesforce.

SF_USERNAME=...  # required
SF_PASSWORD=...  # required
SF_TOKEN=...     # security token
SF_DOMAIN=...    # e.g. 'login' or 'test'

SharePoint / Microsoft Graph

Create an Azure AD app registration that can access the SharePoint site and provide the following values. They are passed directly into SharePointClient.

SHAREPOINT_TENANT_ID=...
SHAREPOINT_CLIENT_ID=...
SHAREPOINT_CLIENT_SECRET=...
SHAREPOINT_SITE_HOSTNAME=brahmcentre.sharepoint.com
SHAREPOINT_SITE_PATH=/sites/BrahmCentre

Usage

Quick import

from brahm_centre_utils import (
    query_salesforce,
    sf_create_from_df,
    send_email_via_rest_api,
    SharePointClient,
)

Salesforce helpers

from simple_salesforce import Salesforce
from brahm_centre_utils import (
    query_salesforce,
    get_picklist_options,
    sf_create_from_df,
    sf_update_from_df,
    sf_delete_from_df,
    sf_upsert_from_df,
    send_email_via_rest_api,
)

sf = Salesforce(
    username=os.getenv("SF_USERNAME"),
    password=os.getenv("SF_PASSWORD"),
    security_token=os.getenv("SF_TOKEN"),
    domain=os.getenv("SF_DOMAIN", "login"),
)

# Query data as a flattened DataFrame
accounts = query_salesforce("SELECT Id, Name, Industry FROM Account", sf)

# Inspect metadata
industries = get_picklist_options("Account", "Industry", sf)

# Bulk create/update/delete from pandas DataFrames
sf_create_from_df(sf, new_records_df, "CustomObject__c")
sf_update_from_df(sf, updates_df, "CustomObject__c")  # requires Id column
sf_delete_from_df(sf, deletions_df, "CustomObject__c")
sf_upsert_from_df(sf, upserts_df, "CustomObject__c", external_id_field="LegacyId__c")

# Send an email via REST API
send_email_via_rest_api(
    sf,
    ["[e-mail]"],
    subject="Automation Update",
    body="<p>Your files have been processed.</p>",
)

Other Salesforce utilities:

• get_Id_from_SF_URL(url) – split a Salesforce record Id from a URL.
• send_coa_certificates(query, sf) – toggle certificate flags for Campaign Members.

SharePoint helpers

from brahm_centre_utils import SharePointClient

client = SharePointClient(
    tenant_id=os.getenv("SHAREPOINT_TENANT_ID"),
    client_id=os.getenv("SHAREPOINT_CLIENT_ID"),
    client_secret=os.getenv("SHAREPOINT_CLIENT_SECRET"),
    site_hostname=os.getenv("SHAREPOINT_SITE_HOSTNAME"),
    site_path=os.getenv("SHAREPOINT_SITE_PATH"),
)

# Upload a bytes buffer or file
client.upload_file("Shared Documents/Reports/output.xlsx", my_bytes)

# List items sorted by last modified date
files = client.list_files_in_folder("Shared Documents/Reports")

# Download structured data
df = client.get_csv_as_df("Shared Documents/Data/latest.csv")
sheet = client.get_xlsx_as_df("Shared Documents/Data/data.xlsx", sheet_name="Sheet1")

# Other helpers
client.create_folder("Shared Documents/Archive/2024")
client.move_file(
    "Shared Documents/Reports/output.xlsx",
    "Shared Documents/Archive/output_2024.xlsx",
)
client.get_list_dataframe("Daily Updates")
client.find_item_by_name("output.xlsx")

Data-cleaning helpers

from brahm_centre_utils import find_contact_duplicates

deduped = find_contact_duplicates(df, name_column="Full Name")

Testing

pytest

The development extras include pytest so the above command works after pip install -e .[dev].

Publishing (internal process)

1. Update the version number in setup.py.
2. Build distributions:

   python setup.py sdist bdist_wheel
   

3. Upload to PyPI (or the internal index):

   twine upload dist/*