APIs integrations and Database crud ops in clean architecture design
In this article, I want to discuss a bit more on supporting API integrations along with DB ops for a DDD codebase which leverages on top of the clean architecture philosophy. This is going to be a continuation of my previous post https://medium.com/@surajit.das0320/understanding-clean-architecture-in-python-deep-dive-on-the-code-17141dc5761a
I will specifically dive into the repository area and touch a bit on the use case layer. Let’s dive in -
So, you have your DDD project set up for supporting customer records. You have your CRUD ops which does all the Insert, Update, Get and Delete. Well, great. This is how the current code looks like
Let’s call our app customer.
customer
repo
__init__.py
base.py
customer.pySo, we see that we have the repo and it has two modules for now.
We have the base module where we define the AbstractCustomerRepository Interface. e.g
class AbstractCustomerRepository(ABC):
@abstractmethod
def insert(self, customer: CustomerEntity) -> Optional[CustomerEntity]:
...
@abstractmethod
def update(self, customer: CustomerEntity) -> CustomerEntity:
...
@abstractmethod
def get_by_id(self, customer_id) -> Optional[CustomerEntity]:
...
@abstractmethod
def delete(self, customer_id):
...
@abstractmethod
def list(self) -> Optional[Sequence[CustomerEntity]]:
...Then we have the customer.py, where we do all the implementation for the CustomerRepository interface. e.g
class CustomerRepository(AbstractCustomerRepository):
def insert(self, obj: CustomerEntity) -> CustomerEntity:
customer = Customer.objects.create(**obj.dict())
return CustomerEntity(**customer.__dict__)
def update(self, obj: CustomerEntity):
Customer.objects.filter(id=obj.id).update(**obj.dict())
customer = Customer.objects.get(id=obj.id)
return CustomerEntity(**customer.__dict__)
def get_by_id(self, customer_id) -> Optional[CustomerEntity]:
if customer := Customer.objects.filter(id=customer_id).first():
return CustomerEntity(**customer.__dict__)
else:
return None
def delete(self, customer_id) -> None:
Customer.objects.get(id=customer_id).delete()
def list(self) -> Optional[Sequence[CustomerEntity]]:
customers = Customer.objects.all()
return [CustomerEntity(**customer.__dict__) for customer in customers]As you can see this specific implementation is very DB centric which serves well for our purpose till now.
Now, let’s say you have a new requirement, the incoming customer data needs to pushed to an external CRM let’s say Hubspot. What would be the way to go about it?
So, in the existing codebase, we are saving the customer data in our Database which is a part of our infrastructure and then enter Hubspot. Good thing about DDD design is that it is made for such kind of requirements where your code needs to be infrastructure agnostic.
So, let’s do a slight restructure for the repo layer
How about this ?
customer
repo
db/
__init__.py
base.py
customer.py
api/
__init__.py
base.py
hubspot_api.pyOk, so what did we do here ? We restructured the repo and add one more layer which is db and api.
All the database related ops will sit inside the db and the external apis will sit inside the api. Neat. Let’s do a quick implementation
customer/api/base.py
from abc import abstractmethod, ABC
from typing import Optional, Sequence
from core.apps.crm.domain.entities import HubspotCustomer
class AbstractHubspotAPICustomerRepository(ABC):
@abstractmethod
def insert(self, customer: HubspotCustomer) -> Optional[HubspotCustomer]:
...
@abstractmethod
def update(self, customer: HubspotCustomer) -> Optional[HubspotCustomer]:
...
@abstractmethod
def get_by_id(self, customer_id) -> Optional[HubspotCustomer]:
...
@abstractmethod
def delete(self, customer_id):
...
@abstractmethod
def list(self) -> Optional[Sequence[HubspotCustomer]]:
...Few changes
- The abstract repository has been prefixed with outgoing client’s name. In this case Hubspot.
- The customer type for the parameter has been modified to HubspotCustomerEntity. Why? Because, hubspot takes in data differently than our own domain structure.
- Same with output data is of type HubspotCustomerEntity which is because Hubspot gives us the response in a different format than our domain response.
So, we have our domain entity which looks like this
customer/domain/entities.py
class Customer(BaseModel):
id: uuid.UUID = Field(default_factory=lambda: uuid.uuid4())
first_name: str
last_name: str
email: EmailStr
phone: str
@validator("phone")
def phone_validation(cls, v):
# For reference - https://stackoverflow.com/questions/70414211/pydantic-custom-data-type-for-phone-number
# -value-error-missing
regex = r"^(\+)[1-9][0-9\-\(\)\.]{9,15}$"
if v and not re.search(regex, v, re.I):
raise InvalidPhoneNumberException("Invalid Phone Number.")
return vcustomer/domain/entities.py
class HubspotCustomer(Entity):
firstname: str
lastname: str
email: EmailStr
phone: str
company: str
@validator("phone")
def phone_validation(cls, v):
# For reference - https://stackoverflow.com/questions/70414211/pydantic-custom-data-type-for-phone-number
# -value-error-missing
regex = r"^(\+)[1-9][0-9\-\(\)\.]{9,15}$"
if v and not re.search(regex, v, re.I):
raise InvalidPhoneNumberException("Invalid Phone Number.")
return vcustomer/repo/api/exceptions.py
class HubspotAPIException(Exception):
...
class ContactException(HubspotAPIException):
...customer/repo/api/hubspot_api.py
pip install --upgrade hubspot-api-client
from typing import Optional
from hubspot.hubspot import HubSpot
from hubspot.crm.contacts import SimplePublicObjectInput
from hubspot.crm import contacts
from django.conf import settings
from core.apps.crm.domain.entities import HubspotCustomer
from core.apps.crm.repo.api.base import AbstractHubspotAPICustomerRepository
from core.apps.crm.repo.api.exceptions import ContactException
class HubspotAPICustomerRepository(AbstractHubspotAPICustomerRepository):
api_client = HubSpot(access_token=settings.HUBSPOT_API_KEY)
def insert(self, customer: HubspotCustomer) -> Optional[HubspotCustomer]:
try:
simple_public_object_input = SimplePublicObjectInput(properties=customer.dict())
response = self.api_client.crm.contacts.basic_api.create(
simple_public_object_input=simple_public_object_input
)
customer = HubspotCustomer(**response.properties)
return customer
except contacts.exceptions.ApiException as e:
raise ContactException(e)So, we have done a basic integration with Hubspot CRM. We implemented the HubspotAPICustomerRepository and implemented the insert operation. A sample Hubspot response for creating a contact i.e customer looks like the following —
{'archived': False,
'archived_at': None,
'created_at': datetime.datetime(2023, 5, 7, 14, 2, 52, 386000, tzinfo=tzutc()),
'id': '201',
'properties': {'company': 'HubSpot',
'createdate': '2023-05-07T14:02:52.386Z',
'email': 'abc@example1.com',
'firstname': 'test1',
'hs_all_contact_vids': '201',
'hs_email_domain': 'example1.com',
'hs_is_contact': 'true',
'hs_is_unworked': 'true',
'hs_lifecyclestage_lead_date': '2023-05-07T14:02:52.386Z',
'hs_object_id': '201',
'hs_pipeline': 'contacts-lifecycle-pipeline',
'hs_searchable_calculated_phone_number': '5551222423',
'lastmodifieddate': '2023-05-07T14:02:52.386Z',
'lastname': 'testerson1',
'lifecyclestage': 'lead',
'phone': '555-122-2423'},
'properties_with_history': None,
'updated_at': datetime.datetime(2023, 5, 7, 14, 2, 52, 386000, tzinfo=tzutc())}In this article, we went over how we can support external API integrations along with supporting Database operations in a clean architecture.
This is just a sample approach. In my next article, I will explain how the data transformation will work in the use case layer to support the above operations.
