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

Compare commits

base: ruben:a95dfaae2becb925b88fa72dad019763aa52d4ad
Branches Tags
ruben:main
..
compare: ruben:38416f103f3216e7e75315c050bc228aa648d9fc
Branches Tags
ruben:main

No commits in common. "a95dfaae2becb925b88fa72dad019763aa52d4ad" and "38416f103f3216e7e75315c050bc228aa648d9fc" have entirely different histories.
a95dfaae2b ... 38416f103f

1 changed files with 11 additions and 22 deletions
Show Stats Expand all files Collapse all files

33
app/main.py
View File

@ -15,13 +15,11 @@ 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): class IPAddressParam(BaseModel):
''' '''
Payload entry as used in POST Payload entry as used in POST
''' '''
ip: Union[IPv6Address, IPv4Address] ip: Union[IPv6Address,IPv4Address]
class Locality(BaseModel): class Locality(BaseModel):
''' '''
@ -32,24 +30,20 @@ class Locality(BaseModel):
continent: Optional[str] = None continent: Optional[str] = None
is_eu: bool = False is_eu: bool = False
class GeoLocation(BaseModel): class GeoLocation(BaseModel):
''' '''
Geolocation data model Geolocation data model
''' '''
ip: Optional[Union[IPv6Address, IPv4Address]] = None ip: Optional[Union[IPv6Address,IPv4Address]] = None
asn: Optional[int] = None asn: Optional[int] = None
asn_org: Optional[str] = None asn_org: Optional[str] = None
network: Optional[Union[IPv6Network, IPv4Network]] = None network: Optional[Union[IPv6Network,IPv4Network]] = None
locality: Locality = Locality() locality: Locality = Locality()
@app.post("/") @app.post("/")
async def root_post(ipaddresses: Annotated[ async def root_post(ipaddresses: Annotated[list[IPAddressParam],
list[IPAddressParam], Body(title="The IPAddresses to geolocate")]
Body(title="The IPAddresses to geolocate")], ) -> list[GeoLocation]:
response: Response
) -> list[GeoLocation]:
''' '''
Return GeoLocation item(s) for a list of IPAddressParam objects Return GeoLocation item(s) for a list of IPAddressParam objects
''' '''
@ -78,31 +72,26 @@ async def root_post(ipaddresses: Annotated[
ip=ipaddress.ip ip=ipaddress.ip
) )
) )
if geolocations:
response.headers['Cache-Control'] = 'private, max-age=604800'
return geolocations return geolocations
@app.get("/{ipaddress}") @app.get("/{ipaddress}")
async def root_get(ipaddress: Annotated[ async def root_get(ipaddress: Annotated[Union[IPv4Address,IPv6Address],
Union[IPv4Address, IPv6Address], Path(title="The IPAddress to geolocate")],
Path(title="The IPAddress to geolocate")],
response: Response response: Response
) -> GeoLocation: ) -> GeoLocation:
''' '''
Look up geolocation for ip in path parameter Look up geolocation for ip in path parameter
''' '''
locations = await root_post([IPAddressParam(ip=ipaddress)], response) locations = await root_post([IPAddressParam(ip=ipaddress)])
if locations: if locations:
response.headers['Cache-Control'] = 'private, max-age=604800'
return locations.pop() return locations.pop()
response.status_code = status.HTTP_404_NOT_FOUND response.status_code = status.HTTP_404_NOT_FOUND
return GeoLocation() return GeoLocation()
@app.get("/") @app.get("/")
def root_redirect(req: Request) -> RedirectResponse: def root_redirect(req: Request) -> RedirectResponse:
''' '''
Redirect empty request using REMOTE_ADDR Redirect empty request using REMOTE_ADDR
''' '''
return RedirectResponse(url=str(req.url) + str(req.client.host)) return RedirectResponse(url= str(req.url) + str(req.client.host))