ruben/ismijnipverweg
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Handle AddressNotFoundError and allow empty placeholder GeoLocation for failed queries

Browse Source
This commit is contained in:
Ruben van Staveren 2023-12-18 14:59:27 +01:00
parent 949e8582bc
commit 0b4322d671
Signed by: ruben
GPG Key ID: 886F6BECD477A93F
1 changed files with 38 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
app/main.py
View File

@ -5,7 +5,8 @@ from ipaddress import IPv4Address, IPv4Network, IPv6Address, IPv6Network
from typing import Annotated, Optional, Union from typing import Annotated, Optional, Union
import geoip2.database import geoip2.database
from fastapi import FastAPI, Path, Body from geoip2.errors import AddressNotFoundError
from fastapi import FastAPI, Path, Body, Response, status
from pydantic import BaseModel from pydantic import BaseModel
app = FastAPI() app = FastAPI()
@ -23,20 +24,20 @@ class Locality(BaseModel):
''' '''
Locality data Locality data
''' '''
city: Optional[str] city: Optional[str] = None
country: Optional[str] country: Optional[str] = None
continent: Optional[str] continent: Optional[str] = None
is_eu: bool is_eu: bool = False
class GeoLocation(BaseModel): class GeoLocation(BaseModel):
''' '''
Geolocation data model Geolocation data model
''' '''
ip: Union[IPv6Address,IPv4Address] ip: Optional[Union[IPv6Address,IPv4Address]] = None
asn: Optional[int] asn: Optional[int] = None
asn_org: Optional[str] asn_org: Optional[str] = None
network: Union[IPv6Network,IPv4Network,None] network: Optional[Union[IPv6Network,IPv4Network]] = None
locality: Locality locality: Locality = Locality()
@app.post("/") @app.post("/")
async def root_post(ipaddresses: Annotated[list[IPAddressParam], async def root_post(ipaddresses: Annotated[list[IPAddressParam],
@ -49,28 +50,40 @@ async def root_post(ipaddresses: Annotated[list[IPAddressParam],
with (geoip2.database.Reader(GEOLITE2_ASN_DB) as reader_asn, with (geoip2.database.Reader(GEOLITE2_ASN_DB) as reader_asn,
geoip2.database.Reader(GEOLITE2_CITY_DB) as reader_city): geoip2.database.Reader(GEOLITE2_CITY_DB) as reader_city):
for ipaddress in ipaddresses: for ipaddress in ipaddresses:
asn_data = reader_asn.asn(ipaddress.ip) try:
city_data = reader_city.city(ipaddress.ip) asn_data = reader_asn.asn(ipaddress.ip)
geolocations.append(GeoLocation( city_data = reader_city.city(ipaddress.ip)
ip=ipaddress.ip,
asn=asn_data.autonomous_system_number, geolocations.append(GeoLocation(
asn_org=asn_data.autonomous_system_organization, ip=ipaddress.ip,
network=asn_data.network, asn=asn_data.autonomous_system_number,
locality=Locality( asn_org=asn_data.autonomous_system_organization,
city=city_data.city.name, network=asn_data.network,
country=city_data.country.iso_code, locality=Locality(
continent=city_data.continent.code, city=city_data.city.name,
is_eu=city_data.country.is_in_european_union country=city_data.country.iso_code,
continent=city_data.continent.code,
is_eu=city_data.country.is_in_european_union
)
))
except AddressNotFoundError:
geolocations.append(GeoLocation(
ip=ipaddress.ip
)
) )
))
return geolocations return geolocations
@app.get("/{ipaddress}") @app.get("/{ipaddress}")
async def root_get(ipaddress: Annotated[Union[IPv4Address,IPv6Address], async def root_get(ipaddress: Annotated[Union[IPv4Address,IPv6Address],
Path(title="The IPAddress to geolocate")] Path(title="The IPAddress to geolocate")],
response: Response
) -> GeoLocation: ) -> GeoLocation:
''' '''
Look up geolocation for ip in path parameter Look up geolocation for ip in path parameter
''' '''
return (await root_post([IPAddressParam(ip=ipaddress)])).pop() locations = await root_post([IPAddressParam(ip=ipaddress)])
if locations:
return locations.pop()
response.status_code = status.HTTP_404_NOT_FOUND
return GeoLocation()