From 478e25036852eb31b7e0a810d6e35135aaa07c19 Mon Sep 17 00:00:00 2001
From: egutierrez <egutierrez@dead.dd>
Date: Fri, 29 Nov 2024 00:58:25 +0100
Subject: [PATCH] Add initial implementation for weather data retrieval and
 storage

- Create .env file for environment variables
- Add .gitignore to exclude virtual environment
- Implement weatherstack.py to fetch weather data from API and store it in MongoDB
- Define tasks for data retrieval, processing, and storage using Prefect
---
 .env                                     |   5 ++
 .gitignore                               |   1 +
 __pycache__/weatherstack.cpython-312.pyc | Bin 0 -> 5103 bytes
 weatherstack.py                          |  97 +++++++++++++++++++++++
 4 files changed, 103 insertions(+)
 create mode 100644 .env
 create mode 100644 .gitignore
 create mode 100644 __pycache__/weatherstack.cpython-312.pyc
 create mode 100644 weatherstack.py

diff --git a/.env b/.env
new file mode 100644
index 0000000..633e31a
--- /dev/null
+++ b/.env
@@ -0,0 +1,5 @@
+API_URL=http://api.weatherstack.com/current
+API_KEY=d46d7419401ae4d339037fd5be77e5d7
+MONGO_URI=mongodb://admin:iO6RQO3xPoXvpm7BXUKZAjhijYLssQ@10.8.0.3:27017/
+DB_NAME=weather_data
+COLLECTION_NAME=current_weather
\ No newline at end of file
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..f5e96db
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1 @@
+venv
\ No newline at end of file
diff --git a/__pycache__/weatherstack.cpython-312.pyc b/__pycache__/weatherstack.cpython-312.pyc
new file mode 100644
index 0000000000000000000000000000000000000000..09f226afcafff08b79af8df0374fd67699fd77d1
GIT binary patch
literal 5103
zcmbuCOKcm*8G!fA<wK+%BxQ?|LRpC{rnO`_ZW1wu9ZHrHC6??oshy|>7HjTGUU|u7
zW|wg&P_7aLsN5c$patxrHsAt1IB*U=>QcZ?F9CWXOBH4}4jQ0Ef!^r4MFSTF`p+zv
z4?R#6S&*~;%=|MukN=zhpWk(K_$YYv|Gc{NlORR?8E4$bQx_ipm8Pg`ltL-APA%X!
zqth8?fuXU?>THHv;4=IIPx72DWIPKVBJ;YK@h*62ilJ1`oqBhqM$mm3|AL>$v>sRp
zfJR`s&;4}st#}mi4&&q(g1c&Y6<<Ruw5yh12{g1ich%}pf(@;%UA00=XG5!7<(00F
zD0hYnJ&^Bi$%i4|(~^&<oDy!yN7Y^>a&eTR&`~wIJkgxJE5RuC!r!m<-GRNp&+_xm
zyES#|l;~%@?%0+mTQwY6*`w@L`ZQXJeb#@6b;@X}Ko{u5z~69oK0(|5IU~DdOzWDO
zwGxcYE$YT4o3mu|BCga8S&<aOQnT+tHn^l(63S;K-B?;u5tMMJW$S|NNu8dRe(?R1
zE5jEoE0-J}mvh?KB^9bvWLk3i;#k_qjHmMm!ABhJ*Js|bLsnh_P3fXqkmPssn(duC
zHUG*f=xKJPFN24roQKYIBQ5I}4AV*;dg0*23*)xu_>?rCnwzl$^>0XuY{_<L`qasj
zGt+O)o|<>6L-j98b+vSTP032|km5CYe3V$d)u?lKca<*E%k@<wE$WFv7pY%THyL*q
zXv*T;`XVrMgIQ&~l;u)IEsJh#^KB6talXiW#=3jB$`;w>j;59?xg(;fT}H&=5h?sl
zwmBx~Dde_qzTKYQq=qT0(=C6RLVc?o^xfl@G~bF)<l1NZA*Fa$xy#%o`V#d%`wn%9
zPKfzAXfbVM&Ae{Oab1psLBtg`ezwI1;yD?~apOGLl8WN6V}=<|>sscwzp}J6$e^Cj
z%JH<8S7c=@{}S{#o7K{qoK+0mW!mi?6t}k~ontN1JQ)X5_Q~nAYMR)*Z2sN6iV9$`
z0#?kJ`Els}__i@1HLac36g)Bmj5Te^8L(K@8Tu1?EdMk#n?cAxaarGeF3I?O!iTU0
z*qoMKG!WiyTSO|1!n91A1&)vxmvtgpGgS#xBml+An>M#>8d=*r^L|>*SsIAEiociO
zZNZtFEjaX{aJ^GA2P>o6v}$6Tj^pRab2G_vr;$;BDVXOlor?Azu&7F86*Hvgb8Up?
z3ih7*=3tx*Q^K1glf6Ovpkr(t<~2yxsDB7lu)oqVuy&%#`F{NR2d|g9=e}6{;_XuB
znN9BOU!numuHHvH8x+<~)I<*-eApAacI1;IzxLb}HhW%J7pr3D197M<4sD9VRWW$g
zf5l&l&fHtP_x2ZWl)|Sr#naVL@72{St3UlwNgVv<p+Ee0is6G;HE<<R>V0V=weegj
za%@w4xhi)2yT(K16LSxY?bB2!HA&r{^iK7&pEHrEUiR}|9^^Lk1yCgD_y6DMHv#hk
z>P6~38u=}}?Br1z0DcIEgu?BGQiaE9BPY~z45{<9<-|5kFVXPb;8&Slu=f|4b~K9&
z0UOVcJOpeOT4P(=@+g5K+uj<$?pS3nvn^l;S4O~!fFghzBT9g2LINZ@tdHdnLibHv
zRXv__+GVg`F)-ZQm>sh{^#gik#6eNZf}hCgDn1%^`id=LUqGnfvVbOZVC1tFD%gHY
z&E!-B1R>Q9H{611rjbUB-L$>Lk62npP4H*{hHG;<fUmWwY<Cf(mlUUA8uJ%iRMQt^
z2OK!vR^rZB>+{=oRwUyq1C0}+u7sMahmQuH91SRn$KX37>v@$lKrtvl{g5O$n<vYK
z2C)uL28H4vo7mN0!qE_}3A!I}kVS`}@ENT38%TEIv~y~co31fIfXN@eK2!D&J?!o;
z4a`)!Pm}^DzK+JOzgCX!zb%xb<JIuq>(7?M&)m|=;e*v!{MO!b>^ryT%CTg1XyjI*
zJTy@{^lEu%wmLX`>$URW_+5W_@W>;-(B)g_YeC8za<G2>ZvO64DSUKOJXQ^b*8|&W
zK799$yVIq}i<{!)7MgFSe5q0D{-}4#&)%PiObP7g0uOQ<&ELY2x6<6C*lO_O)&qfz
z)o|mOOYEwfPMnl*MQ$hOU8EYl@Vju#6!~Qc%C|o4bndLPgCiGqapY`KXzvkt=T?Qw
zLJRNw%KlgKGE!P`Z!vT=8z<4?@u{(un4^fDvqN@k<*>uZ66p&g(yYXOJU)9IFd1%O
z;^?_SK~uNsM+qzlAKEB@6C5R>5G1xw1Lub<$;hg<-;q^C(v%jOJP4mo4od>Cgo98<
zIdA*YhOUE$#FPt2#ledM8TQ;5GI$)$y6Yp9)T~qwOi>b)@ZbCeBs;QpV3QlGF<$V0
zUHv!SsdVjM7pfgSKY!(lw9Zr`y*EPT$mqJS+7)}yHB#;xxs|VU9RL;AAx>_LJV>4_
zCr_3p=St!EP4QGU6nPLDDu;$j;?Q=Ek{c5n!=(rzt=%i$OvV1US8Q<f4b0~n^=$%T
z2!9(svBmzGW>8#ggwLz2)$mTmhW{kt=uW{mvo#zoa1hEi!fE31FlRf2(F$AScOZbj
z%(oG6?1Y|QHo$*@H_~zt=0VWr1fLn#-Ng$K{$=E}3i0Z8;~#1FgIZ42wQN0xY>7};
z#!kCcb+0EY##r6EIED-bI1;V%5_Nb1Jg|GMx5bEfn*M}DYQzl?`)q+gAU@B5z)J>=
z?_PGNBQ_zo0~+8%n1atcJm3S7(BR_We!UqZqGO=*8vM;QNOr=Z`^AmuhFI!MZE{mJ
z21kYA@7_LB2_IhfK8z)9Cn~YSC2`Ng-bCryWTp3|^*{rJ<M+<pn=ggm{Db(`cB?;d
zTfY5fDKfDs9%}I(y)cMd-%CZQ`%!OdlD+>_Bz2g*f0zf^79}YoYgtKJIWdzQKV6TK
z$9IX6$G1eu$?bE8yM|yId6ZUdADk_ykTWzmV)~|WT)vdo(Ds0u#2><(tRPLXqh}3t
z5u$S`mzFGi?~qRB)3TXh(Fjb#_K_=#YJr9wyacLJ&M<Td2SAAdglB4das_fMkrNVf
zNQI!qLO57Jm_%D}?(hijq#df)kKv1imR)jgDxMtN*LCeY!j8^n4AU0ingSPpa#JB!
zm@Q}TR%62^fL9lfOp<K>A~G^kePLMBLbz^G({UZXhT&saP}H1m6h5UMLtBa*wok+(
z8AHkI>QQtS%J9A19D%gP(lq@g75a*L>QB_*SNQkrBY~m=A9Z|1?fH`O)&$BQUf*{$
zaV7EbZ0V`v72nL7r|Roz79FhkCZH$~S)aN(du8_Hi>2XXmB7nu-WILvmn*(8Q0k6-
zyzg4#lSHZiKxzEtN_T24^fk>tVnTHAF-tK*4NJ9sAU$Hb8)YEXMoC#mqYR|lASv@V
z%0R01Q(Rzu<@fB~Di?jg?JINp{ut{1<nVRzXGb6OjFx*wE1^WS=Z7_(;)fqoEHC^^
npgbKPELJ-b)o}my3#Fd@)!y-%2MYg5aUP-OrMNzH2G;ApTgCSs

literal 0
HcmV?d00001

diff --git a/weatherstack.py b/weatherstack.py
new file mode 100644
index 0000000..1013f4c
--- /dev/null
+++ b/weatherstack.py
@@ -0,0 +1,97 @@
+import requests
+from pymongo import MongoClient
+from prefect import flow, task
+from dotenv import load_dotenv
+from prefect.logging import get_run_logger
+
+# Cargar variables de entorno desde un archivo .env
+load_dotenv()
+from pathlib import Path
+import os
+
+# Configuración para la API y MongoDB
+API_URL = os.getenv("API_URL", "http://api.weatherstack.com/current")
+API_KEY = os.getenv("API_KEY", "tu_api_key_aqui")
+MONGO_URI = os.getenv("MONGO_URI", "mongodb://localhost:27017/")
+DB_NAME = os.getenv("DB_NAME", "weather_data")
+COLLECTION_NAME = os.getenv("COLLECTION_NAME", "current_weather")
+
+@task
+def obtener_datos_climaticos(ciudad):
+    """Consulta la API de Weatherstack para obtener datos climáticos de una ciudad."""
+    logger = get_run_logger()
+    logger.info(f"Iniciando la consulta de datos climáticos para la ciudad: {ciudad}")
+    params = {
+        "access_key": API_KEY,
+        "query": ciudad
+    }
+    try:
+        response = requests.get(API_URL, params=params)
+        response.raise_for_status()  # Levanta un error si la solicitud falla
+        logger.info(f"Datos climáticos recibidos correctamente para {ciudad}.")
+        return response.json()
+    except Exception as e:
+        logger.error(f"Error al obtener datos climáticos: {e}")
+        raise
+
+@task
+def procesar_datos(datos):
+    """Procesa los datos recibidos de la API."""
+    logger = get_run_logger()
+    logger.info("Iniciando el procesamiento de datos climáticos.")
+    try:
+        if "current" not in datos:
+            raise ValueError("Datos incompletos recibidos de la API.")
+        clima = {
+            "ciudad": datos["location"]["name"],
+            "pais": datos["location"]["country"],
+            "temperatura": datos["current"]["temperature"],
+            "clima_descripcion": datos["current"]["weather_descriptions"],
+            "fecha_consulta": datos["location"]["localtime"]
+        }
+        logger.info(f"Datos procesados correctamente: {clima}")
+        return clima
+    except Exception as e:
+        logger.error(f"Error al procesar los datos: {e}")
+        raise
+
+@task
+def guardar_en_mongodb(datos):
+    """Guarda los datos procesados en MongoDB."""
+    logger = get_run_logger()
+    logger.info("Guardando datos en MongoDB.")
+    try:
+        client = MongoClient(MONGO_URI)
+        db = client[DB_NAME]
+        collection = db[COLLECTION_NAME]
+        result = collection.insert_one(datos)
+        logger.info(f"Datos guardados en MongoDB con ID: {result.inserted_id}")
+        return result.inserted_id
+    except Exception as e:
+        logger.error(f"Error al guardar datos en MongoDB: {e}")
+        raise
+
+@flow
+def pipeline_climatico(ciudad):
+    """Flujo principal que consume la API y almacena datos en MongoDB."""
+    logger = get_run_logger()
+    logger.info(f"Iniciando el pipeline para la ciudad: {ciudad}")
+    try:
+        datos_api = obtener_datos_climaticos(ciudad)
+        datos_procesados = procesar_datos(datos_api)
+        id_insertado = guardar_en_mongodb(datos_procesados)
+        logger.info(f"Pipeline completado. Datos insertados con ID: {id_insertado}")
+    except Exception as e:
+        logger.error(f"Error en el pipeline: {e}")
+        raise
+
+# Configura y despliega el flujo
+if __name__ == "__main__":
+    pipeline_climatico.from_source(
+        source=str(Path(__file__).parent),  # Código almacenado en el directorio local
+        entrypoint="E:/Proyects/API_diverse_consumption/weatherstack.py:pipeline_climatico",
+    ).deploy(
+        name="Conseguir_datos_climaticos",
+        parameters={"ciudad": "Madrid"},
+        work_pool_name="Workers_pc_torre_Lucas"
+    )