Skip to content

Commit

Permalink
fix(deps): update all patch dependencies (#2104)
Browse files Browse the repository at this point in the history
* fix(deps): update all patch dependencies

* Bump utils version

* Add typing for salesforce client

- Salesforce 1.12.6 introduced typing to the project so we need to conform or mypy gets upset

* formatting

---------

Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com>
Co-authored-by: wbanks <[email protected]>
  • Loading branch information
renovate[bot] and whabanks authored May 21, 2024
1 parent 86fb27b commit d1d851d
Show file tree
Hide file tree
Showing 9 changed files with 185 additions and 283 deletions.
2 changes: 1 addition & 1 deletion app/clients/salesforce/salesforce_account.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ def get_org_name_from_notes(organisation_notes: str, name_index: int = ORG_NOTES
return organisation_notes


def get_account_id_from_name(session: Salesforce, account_name: str, generic_account_id: str) -> Optional[str]:
def get_account_id_from_name(session: Optional[Salesforce], account_name: str, generic_account_id: str) -> Optional[str]:
"""Returns the Account ID for the given Account Name. If no match is found, a generic
Account not found ID is returned.
Expand Down
6 changes: 4 additions & 2 deletions app/clients/salesforce/salesforce_auth.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
from typing import Optional

import requests
from flask import current_app
from simple_salesforce import Salesforce
Expand All @@ -13,7 +15,7 @@ def send(self, *args, **kwargs):
return super().send(*args, **kwargs)


def get_session(client_id: str, username: str, password: str, security_token: str, domain: str) -> Salesforce:
def get_session(client_id: str, username: str, password: str, security_token: str, domain: str) -> Optional[Salesforce]:
"""Return an authenticated Salesforce session
Args:
Expand Down Expand Up @@ -46,7 +48,7 @@ def get_session(client_id: str, username: str, password: str, security_token: st
return session


def end_session(session: Salesforce):
def end_session(session: Optional[Salesforce]):
"""Logout of a Salesforce session
Args:
Expand Down
8 changes: 5 additions & 3 deletions app/clients/salesforce/salesforce_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,15 +28,15 @@ def init_app(self, app):
#
# Authentication
#
def get_session(self) -> Salesforce:
def get_session(self) -> Optional[Salesforce]:
"""Returns an authenticated Salesforce session.
Returns:
Salesforce: The authenticated Salesforce session.
"""
return salesforce_auth.get_session(self.client_id, self.username, self.password, self.security_token, self.domain)

def end_session(self, session: Salesforce) -> None:
def end_session(self, session: Optional[Salesforce]) -> None:
"""Revokes a Salesforce session.
Args:
Expand Down Expand Up @@ -73,7 +73,9 @@ def contact_update(self, user: User) -> None:
salesforce_contact.update(session, user, user_updates)
self.end_session(session)

def contact_update_account_id(self, session: Salesforce, service: Service, user: User) -> Tuple[Optional[str], Optional[str]]:
def contact_update_account_id(
self, session: Optional[Salesforce], service: Service, user: User
) -> Tuple[Optional[str], Optional[str]]:
"""Updates the Account ID for the given Notify user's Salesforce Contact. The Salesforce Account ID
and Contact ID are returned.
Expand Down
12 changes: 6 additions & 6 deletions app/clients/salesforce/salesforce_contact.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
from app.models import User


def create(session: Salesforce, user: User, field_updates: dict[str, Optional[str]]) -> Optional[str]:
def create(session: Optional[Salesforce], user: User, field_updates: dict[str, Optional[str]]) -> Optional[str]:
"""Create a Salesforce Contact from the given Notify User
Args:
Expand All @@ -38,7 +38,7 @@ def create(session: Salesforce, user: User, field_updates: dict[str, Optional[st
"Email": user.email_address,
}
field_values = field_default_values | field_updates
result = session.Contact.create(
result = session.Contact.create( # type: ignore
field_values,
headers={"Sforce-Duplicate-Rule-Header": "allowSave=true"},
)
Expand All @@ -50,7 +50,7 @@ def create(session: Salesforce, user: User, field_updates: dict[str, Optional[st
return contact_id


def update(session: Salesforce, user: User, field_updates: dict[str, Optional[str]]) -> Optional[str]:
def update(session: Optional[Salesforce], user: User, field_updates: dict[str, Optional[str]]) -> Optional[str]:
"""Update a Contact's details. If the Contact does not exist, it is created.
Args:
Expand All @@ -67,8 +67,8 @@ def update(session: Salesforce, user: User, field_updates: dict[str, Optional[st

# Existing contact, update the AccountID
if contact:
result = session.Contact.update(
contact.get("Id"), field_updates, headers={"Sforce-Duplicate-Rule-Header": "allowSave=true"}
result = session.Contact.update( # type:ignore
str(contact.get("Id")), field_updates, headers={"Sforce-Duplicate-Rule-Header": "allowSave=true"}
)
parse_result(result, f"Salesforce Contact update '{user.email_address}' with '{field_updates}'")
contact_id = contact.get("Id")
Expand All @@ -81,7 +81,7 @@ def update(session: Salesforce, user: User, field_updates: dict[str, Optional[st
return contact_id


def get_contact_by_user_id(session: Salesforce, user_id: str) -> Optional[dict[str, str]]:
def get_contact_by_user_id(session: Optional[Salesforce], user_id: str) -> Optional[dict[str, str]]:
"""Retrieve a Salesforce Contact by their Notify user ID. If
they can't be found, `None` is returned.
Expand Down
46 changes: 29 additions & 17 deletions app/clients/salesforce/salesforce_engagement.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from __future__ import annotations

from datetime import datetime
from typing import TYPE_CHECKING, Any, Optional
from typing import TYPE_CHECKING, Any, Dict, Optional

from flask import current_app
from simple_salesforce import Salesforce
Expand All @@ -22,7 +22,11 @@


def create(
session: Salesforce, service: Service, field_updates: dict[str, str], account_id: Optional[str], contact_id: Optional[str]
session: Optional[Salesforce],
service: Service,
field_updates: dict[str, str],
account_id: Optional[str],
contact_id: Optional[str],
) -> Optional[str]:
"""Create a Salesforce Engagement for the given Notify service
Expand All @@ -38,7 +42,7 @@ def create(
"""
engagement_id = None
try:
if account_id and contact_id:
if account_id and contact_id and session:
# Default Engagement values, which can be overridden by passing in field_updates
field_default_values = {
"Name": service.name,
Expand All @@ -54,7 +58,7 @@ def create(
"Product_to_Add__c": ENGAGEMENT_PRODUCT,
}
field_values = field_default_values | field_updates
result = session.Opportunity.create(
result = session.Opportunity.create( # type: ignore
engagement_maxlengths(field_values),
headers={"Sforce-Duplicate-Rule-Header": "allowSave=true"},
)
Expand All @@ -63,7 +67,7 @@ def create(

# Create the Product association
if engagement_id:
result = session.OpportunityLineItem.create(
result = session.OpportunityLineItem.create( # type: ignore
{
"OpportunityId": engagement_id,
"PricebookEntryId": current_app.config["SALESFORCE_ENGAGEMENT_STANDARD_PRICEBOOK_ID"],
Expand All @@ -76,15 +80,19 @@ def create(
parse_result(result, f"Salesforce Engagement OpportunityLineItem create for service ID {service.id}")
else:
current_app.logger.error(
f"SF_ERR Salesforce Engagement create failed: missing Account ID '{account_id}' or Contact ID '{contact_id}' for service ID {service.id}"
f"SF_ERR Salesforce Engagement create failed: missing Account ID '{account_id}' or Contact ID '{contact_id}' for service ID {service.id} or the session is not available. '{session}'"
)
except Exception as ex:
current_app.logger.error(f"SF_ERR Salesforce Engagement create failed: {ex}")
return engagement_id


def update(
session: Salesforce, service: Service, field_updates: dict[str, str], account_id: Optional[str], contact_id: Optional[str]
session: Optional[Salesforce],
service: Service,
field_updates: dict[str, str],
account_id: Optional[str],
contact_id: Optional[str],
) -> Optional[str]:
"""Update an Engagement. If the Engagement does not exist, it is created.
Expand All @@ -104,8 +112,8 @@ def update(

# Existing Engagement, update the stage name
if engagement:
result = session.Opportunity.update(
engagement.get("Id"),
result = session.Opportunity.update( # type: ignore
str(engagement.get("Id")),
engagement_maxlengths(field_updates),
headers={"Sforce-Duplicate-Rule-Header": "allowSave=true"},
)
Expand All @@ -120,7 +128,9 @@ def update(
return engagement_id


def contact_role_add(session: Salesforce, service: Service, account_id: Optional[str], contact_id: Optional[str]) -> None:
def contact_role_add(
session: Optional[Salesforce], service: Service, account_id: Optional[str], contact_id: Optional[str]
) -> None:
"""Adds an Engagement ContactRole based on the provided Notify service and Contact. If the
Engagement does not exist, it is created.
Expand All @@ -136,7 +146,7 @@ def contact_role_add(session: Salesforce, service: Service, account_id: Optional
try:
engagement = get_engagement_by_service_id(session, str(service.id))
if engagement:
result = session.OpportunityContactRole.create(
result = session.OpportunityContactRole.create( # type: ignore
{"ContactId": contact_id, "OpportunityId": engagement.get("Id")},
headers={"Sforce-Duplicate-Rule-Header": "allowSave=true"},
)
Expand All @@ -147,7 +157,9 @@ def contact_role_add(session: Salesforce, service: Service, account_id: Optional
current_app.logger.error(f"SF_ERR Salesforce ContactRole add for {contact_id} with '{service.id}' failed: {ex}")


def contact_role_delete(session: Salesforce, service: Service, account_id: Optional[str], contact_id: Optional[str]) -> None:
def contact_role_delete(
session: Optional[Salesforce], service: Service, account_id: Optional[str], contact_id: Optional[str]
) -> None:
"""Deletes an Engagement ContactRole based on the provided Notify service and Salesforce Contact.
If the Engagement does not exist, it is created.
Expand All @@ -161,19 +173,19 @@ def contact_role_delete(session: Salesforce, service: Service, account_id: Optio
None
"""
try:
result = {}
result: Dict[str, Any] = {}
engagement = get_engagement_by_service_id(session, str(service.id))
engagement_id = engagement.get("Id") if engagement else create(session, service, {}, account_id, contact_id)
engagement_contact_role = get_engagement_contact_role(session, engagement_id, contact_id)

if engagement_contact_role:
result = session.OpportunityContactRole.delete(engagement_contact_role.get("Id"))
result = session.OpportunityContactRole.delete(engagement_contact_role.get("Id")) # type: ignore
parse_result(result, f"Salesforce ContactRole delete for {contact_id} with '{service.id}'")
except Exception as ex:
current_app.logger.error(f"SF_ERR Salesforce ContactRole delete for {contact_id} with '{service.id}' failed: {ex}")


def get_engagement_by_service_id(session: Salesforce, service_id: str) -> Optional[dict[str, Any]]:
def get_engagement_by_service_id(session: Optional[Salesforce], service_id: str) -> Optional[dict[str, Any]]:
"""Retrieve a Salesforce Engagement by a Notify service ID
Args:
Expand All @@ -184,14 +196,14 @@ def get_engagement_by_service_id(session: Salesforce, service_id: str) -> Option
Optional[dict[str, str]]: Salesforce Engagement details or None if can't be found
"""
result = None
if isinstance(service_id, str) and service_id.strip():
if isinstance(service_id, str) and service_id.strip() and session is not None:
query = f"SELECT Id, Name, ContactId, AccountId FROM Opportunity where CDS_Opportunity_Number__c = '{query_param_sanitize(service_id)}' LIMIT 1"
result = query_one(session, query)
return result


def get_engagement_contact_role(
session: Salesforce, engagement_id: Optional[str], contact_id: Optional[str]
session: Optional[Salesforce], engagement_id: Optional[str], contact_id: Optional[str]
) -> Optional[dict[str, Any]]:
"""Retrieve a Salesforce Engagement ContactRole.
Expand Down
13 changes: 8 additions & 5 deletions app/clients/salesforce/salesforce_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ def get_name_parts(full_name: str) -> dict[str, str]:
}


def query_one(session: Salesforce, query: str) -> Optional[dict[str, Any]]:
def query_one(session: Optional[Salesforce], query: str) -> Optional[dict[str, Any]]:
"""Execute an SOQL query that expects to return a single record.
Args:
Expand All @@ -38,11 +38,14 @@ def query_one(session: Salesforce, query: str) -> Optional[dict[str, Any]]:
"""
result = None
try:
results = session.query(query)
if results.get("totalSize") == 1:
result = results.get("records")[0]
if session is not None:
results = session.query(query)
if results.get("totalSize") == 1:
result = results.get("records")[0]
else:
current_app.logger.warn(f"SF_WARN Salesforce no results found for query {query}")
else:
current_app.logger.warn(f"SF_WARN Salesforce no results found for query {query}")
current_app.logger.error("SF_ERR Salesforce session is None")
except Exception as ex:
current_app.logger.error(f"SF_ERR Salesforce query {query} failed: {ex}")
return result
Expand Down
Loading

0 comments on commit d1d851d

Please sign in to comment.