update logic and new method for auth_cognito

app/core/aws.py

@@ -0,0 +1,20 @@
+import boto3
+from dotenv import dotenv_values
+from fastapi import UploadFile
+
+from app.util.file_util import write_file_env
+
+
+def _create_session(config: dict):
+    return boto3.Session(
+        aws_access_key_id=config['AWS_ACCESS_KEY_ID'],
+        aws_secret_access_key=config['AWS_SECRET_ACCESS_KEY'],
+        aws_session_token=config['AWS_SESSION_TOKEN'],
+        region_name=config['AWS_REGION'],
+    )
+
+
+def get_session(path: str, file: UploadFile):
+    write_file_env(path, file)
+    config = dotenv_values(dotenv_path=path)
+    return _create_session(config)

app/core/cognito.py

@@ -0,0 +1,33 @@
+from botocore.client import BaseClient
+from botocore.exceptions import BotoCoreError, ClientError
+from dotenv import dotenv_values
+from fastapi import UploadFile
+
+from app.core.aws import get_session
+
+_PATH_ENV = "cognito.env"
+
+
+class Cognito:
+    @staticmethod
+    def client(file: UploadFile) -> BaseClient:
+        session = get_session(_PATH_ENV, file)
+        return session.client("cognito-idp")
+
+    @staticmethod
+    def auth(client: BaseClient, username: str, password: str):
+        config = dotenv_values(dotenv_path=_PATH_ENV)
+        try:
+            return client.initiate_auth(
+                ClientId=config["AWS_CLIENT_ID_COGNITO"],
+                AuthFlow='USER_PASSWORD_AUTH',
+                AuthParameters={
+                    'USERNAME': username,
+                    'PASSWORD': password
+                },
+            )
+        except (BotoCoreError, ClientError) as error:
+            if isinstance(error, ClientError):
+                return error.response
+            else:
+                return {"Error": error}

app/core/s3.py

@@ -1,22 +1,12 @@
-import os
-import shutil
-
-import boto3
-from dotenv import load_dotenv
 from fastapi import UploadFile
 
+from app.core.aws import get_session
+
+_PATH_ENV = "s3.env"
+
 
-def s3(file: UploadFile):
-    env_path = "s3.env"
-    if os.path.isfile(env_path):
-        os.remove(env_path)
-    with open(env_path, 'w+b') as f:
-        shutil.copyfileobj(file.file, f)
-    load_dotenv(dotenv_path=env_path)
-    session = boto3.Session(
-        aws_access_key_id=os.getenv('AWS_ACCESS_KEY_ID'),
-        aws_secret_access_key=os.getenv('AWS_SECRET_ACCESS_KEY'),
-        aws_session_token=os.getenv('AWS_SESSION_TOKEN'),
-        region_name='us-east-1'
-    )
-    return session.client("s3")
+class S3:
+    @staticmethod
+    def client(file: UploadFile):
+        session = get_session(_PATH_ENV, file)
+        return session.client("s3")

app/model/credentils.py

@@ -1,7 +0,0 @@
-from pydantic import BaseModel
-
-
-class Credentials(BaseModel):
-    aws_access_key_id: str
-    aws_secret_access_key: str
-    aws_session_token: str

app/util/file_util.py

@@ -0,0 +1,10 @@
+import shutil
+
+from fastapi import UploadFile
+
+
+def write_file_env(path: str, file: UploadFile):
+    # if os.path.isfile(env_path):
+    #     os.remove(env_path)
+    with open(path, 'w+b') as f:
+        shutil.copyfileobj(file.file, f)

main.py

@@ -1,9 +1,11 @@
 import os
+from typing import Annotated
 
-from fastapi import FastAPI, File, UploadFile
+from fastapi import FastAPI, File, UploadFile, Form
 from ttp import ttp
 
-from app.core.s3 import s3
+from app.core.cognito import Cognito
+from app.core.s3 import S3
 from app.core.supabase import supabase_client
 from app.core.template import Template
 
@@ -11,20 +13,30 @@ app = FastAPI()
 
 
 @app.post("/data")
-def data(bucket: str, project: str, local_date: str, env_file: UploadFile = File(...)):
+def data(
+        bucket: str,
+        project: str,
+        local_date: str,
+        env_file: UploadFile = File(...)
+):
     server_folder = f"{project}/{local_date}"
     local_folder = f"data/{server_folder}"
-    response = s3(env_file).list_objects_v2(Bucket=bucket, Prefix=server_folder)
+    s3 = S3.client(file=env_file)
+    response = s3.list_objects_v2(Bucket=bucket, Prefix=server_folder)
     if not os.path.exists(local_folder):
         os.makedirs(local_folder)
     with open(f"{local_folder}/data.txt", "ab") as f:
         for obj in response.get('Contents', []):
-            s3(env_file).download_fileobj(bucket, obj['Key'], f)
+            s3.download_fileobj(bucket, obj['Key'], f)
     return {"message": "success"}
 
 
 @app.get("/data")
-async def data(project: str, local_date: str, process: str):
+def data(
+        project: str,
+        local_date: str,
+        process: str
+):
     local_folder = f"data/{project}/{local_date}"
     template = Template.read(process)
     values = {"project": project, "local_date": local_date, "process": process}
@@ -41,3 +53,18 @@ async def data(project: str, local_date: str, process: str):
         except Exception as e:
             print(f"error desde {i}: {e}")
     return result
+
+
+@app.post("/auth")
+def auth(
+        username: Annotated[str, Form()],
+        password: Annotated[str, Form()],
+        env_file: UploadFile = File(...)
+):
+    cognito = Cognito.client(file=env_file)
+    response = Cognito.auth(
+        client=cognito,
+        username=username,
+        password=password
+    )
+    return response

requirements.txt

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:38a92b404f75f1646f77b70e17561f1ebd7802b72eba7f013cb40bb85ecc006e
-size 144
+oid sha256:bea22f86f1eca5015436c9ac7ba1d4dd13e93b594111917741be549cd7dc8376
+size 163