Support for Using Predefined Auth Token from Environment Variables

README.md

@@ -6,6 +6,11 @@
 
 Client for the Quienesquien list service (https://app.q-detect.com/)
 
+> [!IMPORTANT]
+> Generating a new authentication token automatically invalidates any previously created tokens.
+> If multiple applications or services are using the same credentials, creating a new token will render the old ones invalid, potentially causing other applications to fail.
+> To avoid issues, reuse an existing token whenever possible by storing it in an environment variable or a secure location.
+
 ## Installation
 
 ```bash
@@ -33,7 +38,22 @@ export QEQ_CLIENT_ID=your_client_id
 export QEQ_SECRET_ID=your_secret_key
 ```
 
+## Token Generation
+
+Before performing searches, you need to create an authentication token using the create_token method:
+
+```python
+from quienesquien import Client
+
+auth_token = await Client.create_token(os.environ['QEQ_CLIENT_ID'], os.environ['QEQ_SECRET_ID'])
+```
+
+You can reuse this token in subsequent requests.
+
 ## Example
+
+Once you have the token, you can perform searches by passing it:
+
 ```python
 import os
 from quienesquien import Client
@@ -45,10 +65,11 @@ from quienesquien.exc import (
     PersonNotFoundError,
 )
 
+auth_token = await Client.create_token(os.environ['QEQ_CLIENT_ID'], os.environ['QEQ_SECRET_ID'])
+
 client = Client(
     os.environ['QEQ_USER'],
-    os.environ['QEQ_CLIENT_ID'],
-    os.environ['QEQ_SECRET_ID'],
+    auth_token,
 )
 
 try:
@@ -72,6 +93,26 @@ except PersonNotFoundError:
     persons = []
 ```
 
+## Environment Variable (Optional)
+
+To simplify usage, you can store the token in an environment variable:
+
+```bash
+export QEQ_AUTH_TOKEN=your_auth_token
+```
+
+Then, instantiate the client using the environment variable:
+
+```python
+import os
+
+client = Client(
+    os.environ['QEQ_USER'],
+    os.environ['QEQ_CLIENT_ID'],
+    os.environ.get('QEQ_AUTH_TOKEN'),
+)
+```
+
 ## Search Parameters
 - `full_name` (str): Full name of the person.
 - `match_score` (int): Minimum match percentage (default: 60).

env.template

@@ -1,3 +1,4 @@
 QEQ_USER=pepito@cuenca.com
 QEQ_CLIENT_ID=123456-1234-1234
-QEQ_SECRET_ID=notsecurepassword
+QEQ_SECRET_ID=notsecurepassword
+QEQ_AUTH_TOKEN=1234567|authtoken...

quienesquien/client.py

@@ -1,6 +1,6 @@
 import datetime as dt
-from dataclasses import dataclass, field
-from typing import Mapping
+from dataclasses import dataclass
+from typing import Any, ClassVar, Mapping
 
 import httpx
 
@@ -20,24 +20,16 @@
 class Client:
     base_url = 'https://app.q-detect.com'
     username: str
-    client_id: str
-    secret_key: str
-    _auth_token: str | None = None
-    _client: httpx.AsyncClient = field(
-        default_factory=httpx.AsyncClient, init=False
-    )
-
-    def _invalidate_auth_token(self) -> None:
-        """Clear the stored authentication token."""
-        self._auth_token = None
+    auth_token: str | None = None
+    _client: ClassVar[httpx.AsyncClient] = httpx.AsyncClient()
 
     async def _make_request(
         self,
         method: str,
         url: str,
         *,
         headers: Mapping[str, str] | None = None,
-        params: Mapping[str, str | int | None] | None = None,
+        params: Mapping[str, Any] | None = None,
     ) -> httpx.Response:
         """Make an HTTP request using an async client."""
         response = await self._client.request(
@@ -46,25 +38,23 @@ async def _make_request(
         response.raise_for_status()
         return response
 
-    async def _fetch_auth_token(self) -> str:
-        """Retrieve authentication token from the API."""
-        if self._auth_token is not None:
-            return self._auth_token
-
-        auth_url = f'{self.base_url}/api/token'
-        params = {'client_id': self.client_id}
+    @classmethod
+    async def create_token(cls, client_id: str, secret_key: str) -> str:
+        """Create a new authentication token."""
+        auth_url = f'{cls.base_url}/api/token'
+        params = {'client_id': client_id}
         headers = {
-            'Authorization': f'Bearer {self.secret_key}',
+            'Authorization': f'Bearer {secret_key}',
             'Accept-Encoding': 'identity',
         }
         try:
-            response = await self._make_request(
+            response = await cls._client.request(
                 'GET', auth_url, headers=headers, params=params
             )
-            self._auth_token = response.text
+            response.raise_for_status()
         except httpx.HTTPStatusError as exc:
             raise QuienEsQuienError(exc.response) from exc
-        return self._auth_token
+        return response.text
 
     async def search(
         self,
@@ -95,41 +85,31 @@ async def search(
             neither is found.
 
         """
-        # Validate search criteria
-        by_name = full_name is not None
-        by_rfc = rfc is not None
-        by_curp = curp is not None
+        assert (
+            self.auth_token
+        ), 'you must create or reuse an already created auth token'
 
-        if not (by_name or by_rfc or by_curp):
+        if not (full_name or rfc or curp):
             raise InvalidSearchCriteriaError
 
-        token = await self._fetch_auth_token()
-
         # Build base URL with required parameters
         search_url = f'{self.base_url}/api/find'
 
-        params: dict[str, str | int | None] = {
-            'client_id': self.client_id,
+        params = {
             'username': self.username,
             'percent': match_score,
+            'name': full_name,
+            'rfc': rfc,
+            'curp': curp,
+            'sex': gender.value if gender else None,
+            'birthday': birthday.strftime('%d/%m/%Y') if birthday else None,
+            'type': search_type.value if search_type is not None else None,
+            'list': ','.join(search_list) if search_list else None,
         }
 
-        if by_name:
-            params['name'] = full_name
-        if rfc:
-            params['rfc'] = rfc
-        if curp:
-            params['curp'] = curp
-        if gender:
-            params['sex'] = gender.value
-        if birthday:
-            params['birthday'] = birthday.strftime('%d/%m/%Y')
-        if search_type is not None:
-            params['type'] = search_type.value
-        if search_list:
-            params['list'] = ','.join(search_list)
-
-        headers = {'Authorization': f'Bearer {token}'}
+        params = {k: v for k, v in params.items() if v is not None}
+
+        headers = {'Authorization': f'Bearer {self.auth_token}'}
 
         try:
             response = await self._make_request(
@@ -138,7 +118,6 @@ async def search(
         except httpx.HTTPStatusError as exc:
             match exc.response.status_code:
                 case 401:
-                    self._invalidate_auth_token()
                     raise InvalidTokenError(exc.response)
                 case 403:
                     raise InsufficientBalanceError(exc.response)
@@ -155,7 +134,6 @@ async def search(
                     'El token proporcionado para realizar esta acción '
                     'es inválido'
                 ):
-                    self._invalidate_auth_token()
                     raise InvalidTokenError(response)
                 case (
                     'Tu plan de consultas ha expirado, por favor '

quienesquien/person.py

@@ -1,4 +1,10 @@
-from pydantic import BaseModel, ConfigDict, Field, model_validator
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    computed_field,
+    model_validator,
+)
 
 
 class Person(BaseModel):
@@ -15,15 +21,24 @@ class Person(BaseModel):
         default=None, alias='FECHA_NACIMIENTO'
     )
     sexo: str | None = Field(default=None, alias='SEXO')
-    metadata: dict = Field(default_factory=dict, alias='METADATA')
 
     model_config = ConfigDict(
         populate_by_name=True,
         extra='allow',
     )
 
+    @computed_field  # type: ignore[misc]
+    @property
+    def peso1(self) -> str:
+        # peso1 is required for backward compatibility with previous version.
+        return str(self.coincidencia)
+
     @model_validator(mode='after')
     def collect_extra_fields(self):
         if self.model_extra:
-            self.metadata.update(self.model_extra)
+            lowercase_extra = {
+                k.lower(): v for k, v in self.model_extra.items()
+            }
+            self.model_extra.clear()
+            self.model_extra.update(lowercase_extra)
         return self

quienesquien/version.py

@@ -1 +1 @@
-__version__ = '1.0.0'
+__version__ = '1.0.1'

tests/cassettes/test_block_account.yaml

@@ -1,49 +1,4 @@
 interactions:
-- request:
-    body: ''
-    headers:
-      Authorization:
-      - DUMMY_AUTHORIZATION
-      accept:
-      - '*/*'
-      accept-encoding:
-      - identity
-      connection:
-      - keep-alive
-      host:
-      - app.q-detect.com
-      user-agent:
-      - python-httpx/0.28.1
-    method: GET
-    uri: https://app.q-detect.com/api/token?client_id=DUMMY_CLIENT_ID
-  response:
-    body:
-      string: DUMMY_RESPONSE_TOKEN
-    headers:
-      Access-Control-Allow-Origin:
-      - '*'
-      Cache-Control:
-      - no-cache, private
-      Connection:
-      - keep-alive
-      Content-Type:
-      - text/html; charset=UTF-8
-      Date:
-      - Thu, 20 Feb 2025 15:49:53 GMT
-      Server:
-      - nginx/1.18.0 (Ubuntu)
-      Set-Cookie: DUMMY_SET_COOKIE
-      Transfer-Encoding:
-      - chunked
-      X-Content-Type-Options:
-      - nosniff
-      X-Frame-Options:
-      - SAMEORIGIN
-      X-XSS-Protection:
-      - 1; mode=block
-    status:
-      code: 200
-      message: OK
 - request:
     body: ''
     headers:
@@ -60,7 +15,7 @@ interactions:
       user-agent:
       - python-httpx/0.28.1
     method: GET
-    uri: https://app.q-detect.com/api/find?client_id=DUMMY_CLIENT_ID&name=Pepito+Cuenca&percent=80&username=DUMMY_USERNAME
+    uri: https://app.q-detect.com/api/find?name=Pepito+Cuenca&percent=80&username=DUMMY_USERNAME
   response:
     body:
       string: "{\"success\":false,\"status\":\"El token proporcionado para realizar