Add company enrichment service (#15)

Author: ebram96

src/abstract_api/__init__.py

@@ -1,6 +1,7 @@
 from typing import Final
 
 from .avatars import Avatars
+from .company_enrichment import CompanyEnrichment
 from .email_validation import EmailValidation
 from .exchange_rates import ExchangeRates
 from .holidays import Holidays
@@ -11,6 +12,7 @@
 
 __all__: Final[list[str]] = [
     "Avatars",
+    "CompanyEnrichment",
     "EmailValidation",
     "ExchangeRates",
     "Holidays",

src/abstract_api/company_enrichment/__init__.py

@@ -0,0 +1,9 @@
+from typing import Final
+
+from .company_enrichment import CompanyEnrichment
+from .company_enrichment_response import CompanyEnrichmentResponse
+
+__all__: Final[list[str]] = [
+    "CompanyEnrichment",
+    "CompanyEnrichmentResponse"
+]

src/abstract_api/company_enrichment/company_enrichment.py

@@ -0,0 +1,115 @@
+from typing import Iterable
+
+from abstract_api.bases import BaseService
+from abstract_api.exceptions import ClientRequestError, ResponseParseError
+
+from .company_enrichment_response import CompanyEnrichmentResponse
+from .response_fields import RESPONSE_FIELDS
+
+
+class CompanyEnrichment(BaseService):
+    """AbstractAPI company enrichment service.
+
+    Used to find a company's details using its domain.
+
+    Attributes:
+        _subdomain: Company enrichment service subdomain.
+    """
+    _subdomain: str = "companyenrichment"
+
+    def __init__(
+        self,
+        *,
+        response_fields: Iterable[str] | None = None,
+        **kwargs
+    ) -> None:
+        """Constructs a CompanyEnrichment.
+
+        Args:
+            response_fields: Selected response fields.
+        """
+        super().__init__(**kwargs)
+        if response_fields is not None:
+            self.response_fields = frozenset(response_fields)
+        else:
+            self.response_fields = RESPONSE_FIELDS
+
+    @staticmethod
+    def _validate_response_fields(response_fields: Iterable[str]) -> None:
+        """Validates whether all the given fields are acceptable.
+
+        Args:
+            response_fields: Selected response fields.
+        """
+        for field in response_fields:
+            if field not in RESPONSE_FIELDS:
+                raise ClientRequestError(
+                    f"Field '{field}' is not a valid response field for "
+                    f"Company Enrichment service."
+                )
+
+    @property
+    def response_fields(self) -> frozenset[str]:
+        """Gets selected response fields."""
+        if self._response_fields:
+            return self._response_fields
+        return RESPONSE_FIELDS
+
+    @response_fields.setter
+    def response_fields(self, fields: Iterable[str]) -> None:
+        """Sets selected response fields."""
+        self._validate_response_fields(fields)
+        self._response_fields = frozenset(fields)
+
+    @staticmethod
+    def _response_fields_as_param(response_fields: Iterable[str]) -> str:
+        """Builds 'fields' URL query parameter.
+
+         Builds a string that contains selected response fields to be used
+         as a URL query parameter.
+
+        Args:
+            response_fields: Selected response fields.
+
+        Returns:
+            Comma-separated string with all selected response fields.
+        """
+        return ",".join(response_fields)
+
+    def check(
+        self,
+        domain: str,
+        fields: Iterable[str] | None = None
+    ) -> CompanyEnrichmentResponse:
+        """Finds a company's details using its domain.
+
+        Args:
+            domain: The domain of the company you want to get data from.
+            fields: Selected response fields.
+
+        Returns:
+            CompanyEnrichmentResponse representing API call response.
+        """
+        if fields:
+            self._validate_response_fields(fields)
+            response_fields = frozenset(fields)
+        else:
+            response_fields = self.response_fields
+
+        # TODO: Handle request errors.
+        response = self._service_request(
+            domain=domain,
+            fields=self._response_fields_as_param(response_fields)
+        )
+
+        try:
+            company_enrichment_response = CompanyEnrichmentResponse(
+                response=response,
+                response_fields=response_fields
+            )
+        except Exception as e:
+            raise ResponseParseError(
+                "Failed to parse response as CompanyEnrichmentResponse"
+            ) from e
+
+        return company_enrichment_response

src/abstract_api/company_enrichment/company_enrichment_response.py

@@ -0,0 +1,66 @@
+from typing import TYPE_CHECKING
+
+import requests.models
+
+from abstract_api.bases import JSONResponse
+
+
+class CompanyEnrichmentResponse(JSONResponse):
+    """Company enrichment service response."""
+
+    def __init__(
+        self,
+        response: requests.models.Response,
+        response_fields: frozenset[str]
+    ) -> None:
+        """Initializes a new CompanyEnrichmentResponse."""
+        super().__init__(response)
+        self._response_fields = response_fields
+        not_in_response = object()
+        for field in response_fields:
+            if TYPE_CHECKING:
+                assert isinstance(self.meta.body_json, dict)
+            value = self.meta.body_json.get(field, not_in_response)
+            # Set property only if field was returned
+            if value is not not_in_response:
+                setattr(self, f"_{field}", value)
+
+    @property
+    def name(self) -> str | None:
+        """The name of the company."""
+        return self._get_response_field("name")
+
+    @property
+    def domain(self) -> str | None:
+        """The domain the company website is hosted on."""
+        return self._get_response_field("domain")
+
+    @property
+    def year_founded(self) -> int | None:
+        """The year the company was founded."""
+        return self._get_response_field("year_founded")
+
+    @property
+    def industry(self) -> str | None:
+        """The industry the company is operating in."""
+        return self._get_response_field("industry")
+
+    @property
+    def employees_count(self) -> str | None:
+        """The approximate number of employees of the company."""
+        return self._get_response_field("employees_count")
+
+    @property
+    def locality(self) -> str | None:
+        """The city or region he company headquarter is based in."""
+        return self._get_response_field("locality")
+
+    @property
+    def country(self) -> str | None:
+        """The country the company is based in."""
+        return self._get_response_field("country")
+
+    @property
+    def linkedin_url(self) -> str | None:
+        """The LinkedIn URL of the company."""
+        return self._get_response_field("linkedin_url")

src/abstract_api/company_enrichment/response_fields.py

@@ -0,0 +1,13 @@
+"""Response fields of company enrichment service endpoint."""
+
+
+RESPONSE_FIELDS: frozenset[str] = frozenset({
+    "name",
+    "domain",
+    "year_founded",
+    "industry",
+    "employees_count",
+    "locality",
+    "country",
+    "linkedin_url"
+})