HTTP et permet accedir a dades a través d'Internet

Introducció

TODO canviar a httpx

Crea un nou projecte http amb Poetry i una depèndencia a urllib3:

$ poetry new http --name app
$ poetry add urllib3

urllib3 és un client HTTP potent i fàcil d'utilitzar per a Python.

Crea un fitxer test.py:

import urllib3

response = urllib3.request("GET", "https://httpbin.org/robots.txt")
assert(response.status == 200)

print(response.data)

Executa l'script:

$ poetry run python test.py 
b'User-agent: *\nDisallow: /deny\n'

Creating requests

Per fer peticions HTTP/S, importa la llibreria urllib3.

import urllib3

Necessitaràs una instància a PoolManager, que gestiona automàticament la connexió de manera segura, així evitem problemes de concurrència, entre d'altres.

http = urllib3.PoolManager()

Aquest codi realitza una petició (request) GET:

import urllib3

# Creating a PoolManager instance for sending requests.
http = urllib3.PoolManager()

# Sending a GET request and getting back response as HTTPResponse object.
resp = http.request("GET", "https://httpbin.org/robots.txt")

# Print the returned data.
print(resp.data)
# b"User-agent: *\nDisallow: /deny\n"

La funció request() retorna un objecte HTTPResponse. Utilitza request() per fer qualsevol verb de petició (POST, PUT...):

import urllib3

http = urllib3.PoolManager()
resp = http.request(
    "POST",
    "https://httpbin.org/post",
    fields={"hello": "world"} #  Add custom form fields
)

print(resp.data)
# b"{\n "form": {\n "hello": "world"\n  }, ... }

Response Content

L'objecte HTTPResponse proporciona l'estat (200, 404...), data (les dades) i headers (atributs de capçalera):

import urllib3

# Making the request (The request function returns HTTPResponse object)
resp = urllib3.request("GET", "https://httpbin.org/ip")

print(resp.status)
# 200
print(resp.data)
# b"{\n  "origin": "104.232.115.37"\n}\n"
print(resp.headers)
# HTTPHeaderDict({"Content-Length": "32", ...})

JSON Content

Si vols retornar les dades en format JSON, usa el mètode json() quan retornis la resposta:

import urllib3

resp = urllib3.request("GET", "https://httpbin.org/ip")

print(resp.json())
# {"origin": "127.0.0.1"}

Binary Content

L'atribut data de la resposta sempre s'estableix en una cadena de bytes que representa el contingut de la resposta:

import urllib3

resp = urllib3.request("GET", "https://httpbin.org/bytes/8")

print(resp.data)
# b"\xaa\xa5H?\x95\xe9\x9b\x11"

Note. For larger responses, it’s sometimes better to stream the response.

Using io Wrappers with Response Content

Sometimes you want to use io.TextIOWrapper or similar objects like a CSV reader directly with HTTPResponse data. Making these two interfaces play nice together requires using the auto_close attribute by setting it to False. By default HTTP responses are closed after reading all bytes, this disables that behavior:

import io
import urllib3

resp = urllib3.request("GET", "https://example.com", preload_content=False)
resp.auto_close = False

for line in io.TextIOWrapper(resp):
    print(line)
# <!doctype html>
# <html>
# <head>
# ....
# </body>
# </html>

Request Data

Headers

You can specify headers as a dictionary in the headers argument in request():

import urllib3

resp = urllib3.request(
    "GET",
    "https://httpbin.org/headers",
    headers={
        "X-Something": "value"
    }
)

print(resp.json()["headers"])
# {"X-Something": "value", ...}

Or you can use the HTTPHeaderDict class to create multi-valued HTTP headers:

import urllib3

# Create an HTTPHeaderDict and add headers
headers = urllib3.HTTPHeaderDict()
headers.add("Accept", "application/json")
headers.add("Accept", "text/plain")

# Make the request using the headers
resp = urllib3.request(
    "GET",
    "https://httpbin.org/headers",
    headers=headers
)

print(resp.json()["headers"])
# {"Accept": "application/json, text/plain", ...}

Cookies

Cookies are specified using the Cookie header with a string containing the ; delimited key-value pairs:

import urllib3

resp = urllib3.request(
    "GET",
    "https://httpbin.org/cookies",
    headers={
        "Cookie": "session=f3efe9db; id=30"
    }
)

print(resp.json())
# {"cookies": {"id": "30", "session": "f3efe9db"}}

Note that the Cookie header will be stripped if the server redirects to a different host.

Cookies provided by the server are stored in the Set-Cookie header:

import urllib3

resp = urllib3.request(
    "GET",
    "https://httpbin.org/cookies/set/session/f3efe9db",
    redirect=False
)

print(resp.headers["Set-Cookie"])
# session=f3efe9db; Path=/

Query Parameters

Per a les requests amb mètodes HTTP GET, HEAD i DELETE, pots enviar els arguments en un diccionari i els enviarà.

import urllib3

resp = urllib3.request(
    "GET",
    "https://httpbin.org/get",
    fields={"arg": "value"}
)

print(resp.json()["args"])
# {"arg": "value"}

Per a peticions POST i PUT, necessites codificar els paràmetres manualment a la URL, per seguretat:

from urllib.parse import urlencode
import urllib3

# Encode the args into url grammar.
encoded_args = urlencode({"arg": "value"})

# Create a URL with args encoded.
url = "https://httpbin.org/post?" + encoded_args
resp = urllib3.request("POST", url)

print(resp.json()["args"])
# {"arg": "value"}

Form Data

For PUT and POST requests, urllib3 will automatically form-encode the dictionary in the fields argument provided to request():

import urllib3

resp = urllib3.request(
    "POST",
    "https://httpbin.org/post",
    fields={"field": "value"}
)

print(resp.json()["form"])
# {"field": "value"}

JSON

Per enviar JSON al cos d'una sol·licitud, proporciona les dades de l'argument json a request() i urllib3 codificarà automàticament les dades utilitzant el mòdul json amb codificació UTF-8.

A més, quan es proporciona json, el Content-Type a les capçaleres s'estableix a application/json si no s'especifica el contrari.

import urllib3

resp = urllib3.request(
    "POST",
    "https://httpbin.org/post",
    json={"attribute": "value"},
    headers={"Content-Type": "application/json"}
)

print(resp.json())
# {'headers': {'Content-Type': 'application/json', ...},
#  'data': '{"attribute":"value"}', 'json': {'attribute': 'value'}, ...}

Files & Binary Data

Per carregar fitxers mitjançant la codificació de diverses parts/dades de formulari, podeu utilitzar el mateix enfocament que les dades de formulari i especificar el camp de fitxer com una tupla de (file_name, file_data):

import urllib3

# Reading the text file from local storage.
with open("example.txt") as fp:
    file_data = fp.read()

# Sending the request.
resp = urllib3.request(
    "POST",
    "https://httpbin.org/post",
    fields={
       "filefield": ("example.txt", file_data),
    }
)

print(resp.json()["files"])
# {"filefield": "..."}

Tot i que especificar el nom del fitxer no és estrictament necessari, es recomana per tal que coincideixi amb el comportament del navegador. També podeu passar un tercer element a la tupla per especificar explícitament el tipus MIME del fitxer:

resp = urllib3.request(
    "POST",
    "https://httpbin.org/post",
    fields={
        "filefield": ("example.txt", file_data, "text/plain"),
    }
)

Per enviar dades binàries en brut, simplement especifiqueu l'argument body. També es recomana establir la capçalera Content-Type:

import urllib3

with open("/home/samad/example.jpg", "rb") as fp:
    binary_data = fp.read()

resp = urllib3.request(
    "POST",
    "https://httpbin.org/post",
    body=binary_data,
    headers={"Content-Type": "image/jpeg"}
)

print(resp.json()["data"])
# data:application/octet-stream;base64,...

Errors & Exceptions

urllib3 gestiona excepcions de baix nivell, per exemple:

import urllib3

try:
    urllib3.request("GET","https://nx.example.com", retries=False)

except urllib3.exceptions.NewConnectionError:
    print("Connection failed.")
# Connection failed.

Consulta urlib3 - exceptions per a gestionar totes les excepcions.

Streaming and I/O

Quan s'utilitza preload_content=True (la configuració predeterminada), el cos de la resposta es llegirà immediatament a la memòria i la connexió HTTP es tornarà a alliberar al grup sense intervenció manual.

Tanmateix, quan es tracta de respostes grans, sovint és millor reproduir el contingut de la resposta mitjançant preload_content=False.

Això ignifica que urllib3 només llegirà des del sòcol (socket) quan es sol·licitin dades.

note

Quan utilitzeu preload_content=False, heu d'alliberar manualment la connexió HTTP al conjunt de connexions perquè es pugui reutilitzar.

Per assegurar-se que la connexió HTTP està en un estat vàlid abans de ser reutilitzada, totes les dades s'han de llegir de la connexió.

Podeu trucar al drain_conn() per llençar les dades no llegides que encara hi ha a la connexió. Aquesta crida no és necessària si les dades ja s'han llegit completament de la resposta.

Després de llegir totes les dades, podeu cridar a release_conn() per alliberar la connexió del Pool de connexions.

Podeu usar close() per tancar la connexió, però aquesta crida no retorna la connexió a la piscina, llença les dades no llegides al cable i deixa la connexió en un estat de protocol indefinit. Això és desitjable si preferiu no llegir les dades del sòcol a reutilitzar la connexió HTTP.

El mètode stream() et permet iterar en trossos d'n bytes el contingut de la resposta.

import urllib3

resp = urllib3.request(
    "GET",
    "https://httpbin.org/bytes/1024",
    preload_content=False
)

for chunk in resp.stream(32):
    print(chunk)
    # b"\x9e\xa97'\x8e\x1eT ....

resp.release_conn()

Tanmateix, també podeu tractar la instància de HTTPResponse com un objecte semblant a un fitxer. Això us permet usar un buffer a la memòria RAM.

import urllib3

resp = urllib3.request(
    "GET",
    "https://httpbin.org/bytes/1024",
    preload_content=False
)

print(resp.read(4))
# b"\x88\x1f\x8b\xe5"

Cridem la funció read() que bloquejarà l'accés fins que arribin més dades a la resposta.

import io
import urllib3

resp = urllib3.request(
    "GET",
    "https://httpbin.org/bytes/1024",
    preload_content=False
)

reader = io.BufferedReader(resp, 8)
print(reader.read(4))
# b"\xbf\x9c\xd6"

resp.release_conn()

Podeu utilitzar aquest objecte semblant a un fitxer per fer coses com descodificar el contingut mitjançant còdecs:

import codecs
import json
import urllib3

reader = codecs.getreader("utf-8")

resp = urllib3.request(
    "GET",
    "https://httpbin.org/ip",
    preload_content=False
)

print(json.load(reader(resp)))
# {"origin": "127.0.0.1"}

resp.release_conn()

Exemple: Baixar un fitxer

import urllib3

url = "https://xtec.dev/xtec.jpg"

http = urllib3.PoolManager()
response = http.request("GET", url)
image_data = response.data
file_name = "xtec.jpg"

with open(file_name, 'wb') as file:
	file.write(image_data)

Cache

Una situació habitual és guardar en cache un fitxer que hem descarregat d'internet.

Per exemple, si estem provant un codi que utilitza la Quantitat d’aigua als embassaments de les Conques Internes de Catalunya més val baixar el fitxer una sola vegada.

import csv
import os.path
import urllib3
    

def get_file(file, url):

    if not os.path.isfile(file):
        response = urllib3.request("GET", url)
        print(response.data)
        with open(file, "wb") as file:
            file.write(response.data)

    return file


def read_csv(file):

    with open(file, mode="r", encoding="utf-8") as file:
        reader = csv.DictReader(file)
        result = [row_dict for row_dict in csv_reader]
        return result
        #for row in reader:
        #    print(row)


file = get_file(
    "aigua.csv",
    "https://analisi.transparenciacatalunya.cat/api/views/gn9e-3qhr/rows.csv?accessType=DOWNLOAD",
)

entries = read_csv(file)

Exemple: Descarregar i obrir un fitxer comprimit.

En aquest cas també apliquem la tècnica de la cache, per a descarregar i descomprimir el fitxer només en cas que no ho haguem fet abans.

import urllib3
import os
import csv
import zipfile

url = "https://gitlab.com/xtec/python/data/-/raw/main/airline-flight.zip?ref_type=heads"
zip_filename = "airline-flight.zip" 

def download_zip(url, zip_filename):
    # El with ens permet tancar automàticament la connexió al acabar.
    with urllib3.PoolManager().request('GET', url, preload_content=False) as response, open(zip_filename, 'wb') as out_file:
        print(f"Descarregant {zip_filename}...")
        while (data := response.read(1024)):  # Utilitzem l'operador d'assignació
            out_file.write(data)
        print(f"Descàrrega completada: {zip_filename}")

def unzip_file_and_get_csv(zip_filename):
    with zipfile.ZipFile(zip_filename, 'r') as zip_ref:
        zip_ref.extractall()  # Descomprimim tots els fitxers
        return next((file for file in zip_ref.namelist() if file.endswith('.csv')), None)

def save_csv_content(csv_filename):
    with open(csv_filename, 'r') as f:
        reader = csv.DictReader(f)
        return [row for row in reader]

def main():
    # Si hi hagués més d'un fitxer
    # csv_filename = next((file for file in os.listdir() if file.endswith('.csv')), None)
    csv_filename = "airline-flight.csv" 
    if not csv_filename:
        if not os.path.exists(zip_filename):
            download_zip(url, zip_filename)
        csv_filename = unzip_file_and_get_csv(zip_filename)

    csv_lines = save_csv_content(csv_filename)
    print(csv_lines[:20]) # Provem si es veu el contingut

if __name__ == "__main__":
    main()

Activitat: Embassaments

1.- Baixa les dades en format Json enlloc de csv.

Pistes:

La URL pel JSON és:

https://analisi.transparenciacatalunya.cat/api/views/gn9e-3qhr/rows.json

Recorda que has d'eliminar les metadades, presents al node 'meta' i que les dades es troben a 'data'.

2.- Llegeix el fitxer Json en un dict Python.

Pista: Si no pots llegir el Json pots llegir el CSV en un dict, però has de ser capaç de transformar qualsevol dels 2.

Per transformar el Json simplement has de seleccionar només les 5 últimes columnes del fitxer CSV.

3.- Analitza les dades:

  • Agafa les dades de l'any 2024, d'un pantà, per exemple el de Riudecanyes.

  • Mostra el dia que hi ha el 'Nivell absolut (msnm)' màxim i el dia mínim.

  • Prova altres pantans, de seleccionar valors concrets del diccionari, altres anys...

from datetime import datetime

# Funció per llegir el fitxer CSV i retornar les línies com una llista de diccionaris
def llegir_csv(nom_fitxer):
    with open(nom_fitxer, 'r') as f:
        lines = f.readlines()
    
    columnes = lines[0].strip().split(',')
    dades = []
    for linia in lines[1:]:
        parts = linia.strip().split(',')
        dades.append({col: val for col, val in zip(columnes, parts)})
    return dades

# Funció per filtrar les dades de l'any 2024
def filtrar_any(dades, any):
    return [d for d in dades if d['Dia'].split('/')[-1] == str(any)]

# Funció per filtrar les dades de Riudecanyes
def filtrar_estacio(dades, estacio):
    return [d for d in dades if estacio in d['Estació']]

# Funció per ordenar les dades per data de més recent a més antic
def ordenar_dades(dades):
    return sorted(dades, key=lambda d: datetime.strptime(d['Dia'], '%d/%m/%Y'), reverse=True)

# Funció per obtenir màxim i mínim
def max_min_percentatge(dades):
    max_dada = dades[0]
    min_dada = dades[0]
    
    for d in dades:
        percentatge = float(d['Nivell absolut (msnm)'])
        if percentatge > float(max_dada['Nivell absolut (msnm)']):
            max_dada = d
        if percentatge < float(min_dada['Nivell absolut (msnm)']):
            min_dada = d
            
    return max_dada, min_dada

# Principal
fitxer = 'aigua.csv'
dades = llegir_csv(fitxer)

# Filtrar les dades
dades_2024 = filtrar_any(dades, 2024)
dades_riudecanyes = filtrar_estacio(dades_2024, 'Riudecanyes')

# Ordenar les dades
dades_ordenades = ordenar_dades(dades_riudecanyes)

# Obtenir màxim i mínim
max_dada, min_dada = max_min_percentatge(dades_ordenades)

# Imprimir resultats
print("Dades de 2024 per Riudecanyes ordenades per data:")
for d in dades_ordenades:
    print(d)

print("\nPercentatge volum embassat màxim:")
print(max_dada)

print("\nPercentatge volum embassat mínim:")
print(min_dada)

Activitat: AEMET

A l'activitat JSON - Objecte vas utilitzar dades de l'AEMET.

Ara anem a crear una aplicació per consultar dades metereològiques.

Instruccions

  1. Ves al projecte https://gitlab.com/xtec/python/aemet
  2. Fes un "fork" del projecte al teu compte de Gitlab
  3. Clona el teu projecte al director arrel
  4. Instal.la les dependències amb poetry update
  5. Executa l'aplicació amb poetry run python app/app.py

Millores

Has de realitzar millores en el projecte.

Per exemple,

  • Dona l'opció d'obtenir altres dades
  • Que l'usuari pugui entra el nom del municipi enlloc del codi.
  • etc.

Desplegament

Puja el teu projecte a DockerHub (veure Registre )

TODO