Compare commits

..

1 Commits

Author SHA1 Message Date
b100d24273
This decorator messes up pydantics BaseModel 2023-12-12 10:50:16 +01:00

View File

@ -1,11 +1,12 @@
''' '''
Simple Geolocation with FastAPI Simple Geolocation with FastAPI
''' '''
import dataclasses
from ipaddress import IPv4Address, IPv4Network, IPv6Address, IPv6Network 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 fastapi import FastAPI, Path
from pydantic import BaseModel from pydantic import BaseModel
app = FastAPI() app = FastAPI()
@ -13,9 +14,6 @@ app = FastAPI()
GEOLITE2_ASN_DB = '/usr/local/share/GeoIP/GeoLite2-ASN.mmdb' GEOLITE2_ASN_DB = '/usr/local/share/GeoIP/GeoLite2-ASN.mmdb'
GEOLITE2_CITY_DB = '/usr/local/share/GeoIP/GeoLite2-City.mmdb' GEOLITE2_CITY_DB = '/usr/local/share/GeoIP/GeoLite2-City.mmdb'
class IPAddressParam(BaseModel):
ip: Union[IPv6Address,IPv4Address]
class Locality(BaseModel): class Locality(BaseModel):
''' '''
Locality data Locality data
@ -35,18 +33,20 @@ class GeoLocation(BaseModel):
network: Union[IPv6Network,IPv4Network,None] network: Union[IPv6Network,IPv4Network,None]
locality: Locality locality: Locality
@app.post("/") @app.get("/{ipaddress}")
async def root_post(ipaddresses: Annotated[list[IPAddressParam], async def root(ipaddress: Annotated[Union[IPv4Address,IPv6Address],
Body(title="The IPAddresses to geolocate")] Path(title="The IPAddress to geolocate")]
) -> list[GeoLocation]: ) -> GeoLocation:
geolocations = [] '''
Look up geolocation for ip in path parameter
'''
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: asn_data = reader_asn.asn(ipaddress)
asn_data = reader_asn.asn(ipaddress.ip) city_data = reader_city.city(ipaddress)
city_data = reader_city.city(ipaddress.ip)
geolocations.append(GeoLocation( return GeoLocation(
ip=ipaddress.ip, ip=ipaddress,
asn=asn_data.autonomous_system_number, asn=asn_data.autonomous_system_number,
asn_org=asn_data.autonomous_system_organization, asn_org=asn_data.autonomous_system_organization,
network=asn_data.network, network=asn_data.network,
@ -56,15 +56,4 @@ async def root_post(ipaddresses: Annotated[list[IPAddressParam],
continent=city_data.continent.code, continent=city_data.continent.code,
is_eu=city_data.country.is_in_european_union is_eu=city_data.country.is_in_european_union
) )
)) )
return geolocations
@app.get("/{ipaddress}")
async def root_get(ipaddress: Annotated[Union[IPv4Address,IPv6Address],
Path(title="The IPAddress to geolocate")]
) -> GeoLocation:
'''
Look up geolocation for ip in path parameter
'''
return (await root_post([IPAddressParam(ip=ipaddress)])).pop()