Version 1.1.2 Added Caching

Author: HMohammad2520

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "py-sarvcrm-api"
-version = "1.1.1"
+version = "1.1.2"
 description = "Simple sarvcrm api module"
 readme = "README.md"
 license = { file = "LICENSE" }
@@ -18,7 +18,8 @@ classifiers = [
     "Operating System :: OS Independent",
 ]
 dependencies = [
-    "requests",
+    "requests==2.32.4",
+    "requests_cache==1.2.1",
 ]
 requires-python = ">=3.9"
 

requirements.txt

@@ -1,4 +1,5 @@
 requests
+requests_cache
 setuptools
 
 python-dotenv

sarvcrm_api/__init__.py

@@ -4,6 +4,7 @@
 from .modules._base import SarvModule
 from .__version__ import __version__ as version
 
+
 __all__ = [
     'SarvClient',
     'SarvURL',

sarvcrm_api/__version__.py

@@ -1,4 +1,4 @@
-__version__ = '1.1.1'
+__version__ = '1.1.2'
 __letter__ = 'v'
 
 def get_version():

sarvcrm_api/_client.py

@@ -1,5 +1,6 @@
-import json, hashlib, requests
-from typing import Any, Dict, List, Optional, Self
+import json, hashlib, requests, requests_cache
+import urllib.parse
+from typing import Any, Dict, List, Literal, Optional, Self
 from datetime import datetime, timedelta, timezone
 from ._exceptions import SarvException
 from ._mixins import ModulesMixin
@@ -21,6 +22,8 @@ def __init__(
             login_type: Optional[str] = None,
             language: SarvLanguageType = 'en_US',
             is_password_md5: bool = False,
+            caching: bool = False,
+            cache_backend: Literal['memory', 'sqlite'] = 'memory',
         ) -> None:
         """
         Initialize the SarvClient.
@@ -34,35 +37,29 @@ def __init__(
             language (SarvLanguageType): The language to use, default is 'en_US'.
             is_password_md5 (bool): Whether the password is already hashed using MD5.
         """
-        self.url = url
-        self.utype = utype
-        self.username = username
-        self.password = password if is_password_md5 else self.hash_password(password)
-        self.login_type = login_type
-        self.language = language
-
-        self.token: str = ''
-        self._session = requests.session()
-        self._session.headers.update({'Content-Type': 'application/json'})
-        self._session.headers.update({'Accept': 'application/json'})
+        self._url = url
+        self._utype = utype
+        self._username = username
+        self._password = password if is_password_md5 else self.hash_password(password)
+        self._login_type = login_type
+        self._language = language
+        self._caching = caching
+        self._cache_backend = cache_backend
 
-        super().__init__()
+        self._token: str = ''
 
+        self.enable_caching() if self._caching else self.disable_caching()
 
-    @staticmethod
-    def hash_password(password: str) -> str:
-        """
-        Returns the acceptable hash for SarvCRM Login
+        super().__init__()
 
-        Args:
-            password(str): your password
-        
-        Returns:
-            str: md5 hashed password
+    def _add_headers(self) -> None:
         """
-        return hashlib.md5(password.encode('utf-8')).hexdigest()
+        Adds required sarvcrm headers to session.
+        """
+        self._session.headers.update({'Content-Type': 'application/json'})
+        self._session.headers.update({'Accept': 'application/json'})
 
-    def create_get_params(
+    def _create_get_params(
             self, 
             sarv_get_method: Optional[SarvGetMethods] = None,
             sarv_module: Optional[SarvModule | str] = None,
@@ -88,7 +85,7 @@ def create_get_params(
                 module_name = sarv_module
             else:
                 raise TypeError(f'Module type must be instance of SarvModule or str not {sarv_module.__class__.__name__}')
-        
+
         get_parms = {
             'method': sarv_get_method,
             'module': module_name,
@@ -100,46 +97,15 @@ def create_get_params(
 
         return get_parms
 
-    @staticmethod
-    def iso_time_output(output_method: TimeOutput, dt: datetime | timedelta) -> str:
-        """
-        Generate a formatted string from a datetime or timedelta object.
-
-        These formats are compliant with the SarvCRM API time standards.
-
-        Args:
-            output_method (TimeOutput): Determines the output format ('date', 'datetime', or 'time').
-            dt (datetime | timedelta): A datetime or timedelta object.
-
-        Returns:
-            str: A string representing the date, datetime, or time.
-                - date: "YYYY-MM-DD"
-                - datetime: "YYYY-MM-DDTHH:MM:SS+HH:MM"
-                - time: "HH:MM:SS"
-        """
-        if isinstance(dt, timedelta):
-            dt = datetime.now(timezone.utc) + dt
-
-        if output_method == 'date':
-            return dt.date().isoformat()
-
-        elif output_method == 'datetime':
-            return dt.astimezone().isoformat(timespec="seconds")
-
-        elif output_method == 'time':
-            return dt.time().isoformat(timespec="seconds")
-
-        else:
-            raise TypeError(f'Invalid output method: {output_method}')
-
-
-    def send_request(
+    def _send_request(
             self, 
             request_method: RequestMethod,
             endpoint: Optional[str] = None,
             head_params: Optional[dict] = None,
             get_params: Optional[dict] = None,
             post_params: Optional[dict] = None,
+            caching: bool = False,
+            expire_after: int = 300,
         ) -> Any:
         """
         Send a request to the Sarv API and return the response data.
@@ -160,17 +126,32 @@ def send_request(
         get_params = get_params or {}
         post_params = post_params or {}
 
-        if self.token:
-            head_params['Authorization'] = f'Bearer {self.token}'
+        if self._token:
+            head_params['Authorization'] = f'Bearer {self._token}'
 
-        response: requests.Response = self._session.request(
-            method = request_method,
-            url = self.url + f'{endpoint if endpoint else ''}',
-            headers = head_params,
-            params = get_params,
-            json = post_params,
-            verify = True,
-        )
+        kwargs = {
+            'method': request_method,
+            'url': urllib.parse.urljoin(self._url.rstrip('/') + '/', (endpoint or '').lstrip('/')),
+            'headers': head_params,
+            'params': get_params,
+            'json': post_params,
+            'verify': True,
+        }
+
+        if isinstance(self._session, requests_cache.CachedSession):
+            ## Use cache
+            if caching:
+                kwargs.update({'expire_after': expire_after})
+                response: requests.Response = self._session.request(**kwargs)
+
+            ## If caching enabled but user dont want to use it
+            else:
+                with self._session.cache_disabled():
+                    response: requests.Response = self._session.request(**kwargs)
+
+        else:
+            ## Normal request without caching
+            response: requests.Response = self._session.request(**kwargs)
 
         # Check for Server respond
         try:
@@ -195,6 +176,26 @@ def send_request(
         return response_dict.get('data', {})
 
 
+    def enable_caching(self) -> None:
+        """
+        Enables the caching and replaces `Session` with `CachedSession` or creates it.
+        """
+        self._session = requests_cache.CachedSession(
+            cache_name='sarv_api_cache',
+            backend=self._cache_backend,
+            allowable_methods=('GET', 'POST'),
+            allowable_codes=(200,),
+        )
+        self._add_headers()
+
+    def disable_caching(self) -> None:
+        """
+        Disables the caching and replaces `CachedSession` with `Session` or creates it.
+        """
+        self._session = requests.Session()
+        self._add_headers()
+
+
     def login(self) -> str:
         """
         Authenticate the user and retrieve an access token.
@@ -203,25 +204,25 @@ def login(self) -> str:
             str: The access token for authenticated requests.
         """
         post_params = {
-            'utype': self.utype,
-            'user_name': self.username,
-            'password': self.password,
-            'login_type': self.login_type,
-            'language': self.language,
+            'utype': self._utype,
+            'user_name': self._username,
+            'password': self._password,
+            'login_type': self._login_type,
+            'language': self._language,
         }
         post_params = {k: v for k, v in post_params.items() if v is not None}
 
-        data: Dict[str, Any] = self.send_request(
+        data: Dict[str, Any] = self._send_request(
             request_method='POST',
-            get_params=self.create_get_params('Login'), 
+            get_params=self._create_get_params('Login'), 
             post_params=post_params,
         )
 
         token = data.get('token', '')
 
         if token is not None:
-            self.token = token
-            return self.token
+            self._token = token
+            return self._token
         else:
             raise SarvException('client did not get token from login request')
 
@@ -231,8 +232,8 @@ def logout(self) -> None:
 
         This method should be called to invalidate the session.
         """
-        if self.token:
-            self.token = ''
+        if self._token:
+            self._token = ''
 
 
     def search_by_number(
@@ -250,19 +251,64 @@ def search_by_number(
         Returns:
             dict: The data related to the phone number if found.
         """
-        return self.send_request(
+        return self._send_request(
             request_method = 'GET',
-            get_params = self.create_get_params(
+            get_params = self._create_get_params(
                 'SearchByNumber', 
                 sarv_module = module, 
                 number = number,
             ),
         )
 
+    @staticmethod
+    def iso_time_output(output_method: TimeOutput, dt: datetime | timedelta) -> str:
+        """
+        Generate a formatted string from a datetime or timedelta object.
+
+        These formats are compliant with the SarvCRM API time standards.
+
+        Args:
+            output_method (TimeOutput): Determines the output format ('date', 'datetime', or 'time').
+            dt (datetime | timedelta): A datetime or timedelta object.
+
+        Returns:
+            str: A string representing the date, datetime, or time.
+                - date: "YYYY-MM-DD"
+                - datetime: "YYYY-MM-DDTHH:MM:SS+HH:MM"
+                - time: "HH:MM:SS"
+        """
+        if isinstance(dt, timedelta):
+            dt = datetime.now(timezone.utc) + dt
+
+        if output_method == 'date':
+            return dt.date().isoformat()
+
+        elif output_method == 'datetime':
+            return dt.astimezone().isoformat(timespec="seconds")
+
+        elif output_method == 'time':
+            return dt.time().isoformat(timespec="seconds")
+
+        else:
+            raise TypeError(f'Invalid output method: {output_method}')
+
+    @staticmethod
+    def hash_password(password: str) -> str:
+        """
+        Returns the acceptable hash for SarvCRM Login
+
+        Args:
+            password(str): your password
+        
+        Returns:
+            str: md5 hashed password
+        """
+        return hashlib.md5(password.encode('utf-8')).hexdigest()
+
 
     def __enter__(self) -> Self:
         """Basic Context Manager for clean code execution"""
-        if not self.token:
+        if not self._token:
             self.login()
 
         return self
@@ -279,7 +325,7 @@ def __repr__(self):
         Returns:
             str: A string containing the class name and key attributes.
         """
-        return f'{self.__class__.__name__}(utype={self.utype}, username={self.username})'
+        return f'{self.__class__.__name__}(utype={self._utype}, username={self._username})'
 
     def __str__(self) -> str:
         """
@@ -288,4 +334,4 @@ def __str__(self) -> str:
         Returns:
             str: A simplified string representation of the instance.
         """
-        return f'<SarvClient {self.utype}-{self.username}>'
+        return f'<SarvClient {self._utype}-{self._username}>'

sarvcrm_api/_mixins.py

@@ -42,4 +42,4 @@ def __init__(self) -> None:
         self.ServiceCenters = ServiceCenters(self)
         self.Tasks = Tasks(self)
         self.Timesheet = Timesheet(self)
-        self.Vendors = Vendors(self)
+        self.Vendors = Vendors(self)

sarvcrm_api/_type_hints.py

@@ -13,6 +13,7 @@
     'SearchByNumber',
 ]
 
+
 __all__ = [
     'TimeOutput',
     'RequestMethod',

sarvcrm_api/_url.py

@@ -5,6 +5,7 @@
 
 SarvURL = SarvAPI_v5
 
+
 __all__ = [
     'SarvAPI_v5',
     'SarvFrontend',