Compare commits

..

1 Commits

Author SHA1 Message Date
b100d24273 This decorator messes up pydantics BaseModel 2023-12-12 10:50:16 +01:00
7 changed files with 30 additions and 156 deletions

View File

@ -1,17 +0,0 @@
---
name: Flake8
on: [push]
# XXX need to do stuff with uv
jobs:
build:
runs-on: freebsd
strategy:
matrix:
python-version: ["3.11"]
steps:
- uses: actions/checkout@v4
- name: Analyse code with Flake8
run: |
flake8 $(git ls-files '*.py')

View File

@ -1,17 +0,0 @@
---
name: Mypy
on: [push]
# XXX need to do stuff with uv
jobs:
build:
runs-on: freebsd
strategy:
matrix:
python-version: ["3.11"]
steps:
- uses: actions/checkout@v4
- name: Analyse code with Mypy
run: |
mypy --install-types --non-interactive $(git ls-files '*.py')

View File

@ -1,17 +0,0 @@
---
name: Pylint
on: [push]
# XXX need to do stuff with uv
jobs:
build:
runs-on: freebsd
strategy:
matrix:
python-version: ["3.11"]
steps:
- uses: actions/checkout@v4
- name: Analyse code with Pylint
run: |
pylint $(git ls-files '*.py')

View File

@ -1,13 +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 geoip2.errors import AddressNotFoundError from fastapi import FastAPI, Path
from fastapi import FastAPI, Path, Body, Request, Response, status
from fastapi.responses import RedirectResponse
from pydantic import BaseModel from pydantic import BaseModel
app = FastAPI() app = FastAPI()
@ -15,54 +14,39 @@ 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):
'''
Payload entry as used in POST
'''
ip: Union[IPv6Address, IPv4Address]
class Locality(BaseModel): class Locality(BaseModel):
''' '''
Locality data Locality data
''' '''
city: Optional[str] = None city: Optional[str]
country: Optional[str] = None country: Optional[str]
continent: Optional[str] = None continent: Optional[str]
is_eu: bool = False is_eu: bool
class GeoLocation(BaseModel): class GeoLocation(BaseModel):
''' '''
Geolocation data model Geolocation data model
''' '''
ip: Optional[Union[IPv6Address, IPv4Address]] = None ip: Union[IPv6Address,IPv4Address]
asn: Optional[int] = None asn: Optional[int]
asn_org: Optional[str] = None asn_org: Optional[str]
network: Optional[Union[IPv6Network, IPv4Network]] = None network: Union[IPv6Network,IPv4Network,None]
locality: Locality = Locality() locality: Locality
@app.get("/{ipaddress}")
@app.post("/") async def root(ipaddress: Annotated[Union[IPv4Address,IPv6Address],
async def root_post(ipaddresses: Annotated[ Path(title="The IPAddress to geolocate")]
list[IPAddressParam], ) -> GeoLocation:
Body(title="The IPAddresses to geolocate")],
response: Response
) -> list[GeoLocation]:
''' '''
Return GeoLocation item(s) for a list of IPAddressParam objects Look up geolocation for ip in path parameter
''' '''
geolocations = []
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)
try: city_data = reader_city.city(ipaddress)
asn_data = reader_asn.asn(ipaddress.ip)
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,
@ -72,37 +56,4 @@ async def root_post(ipaddresses: Annotated[
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
) )
))
except AddressNotFoundError:
geolocations.append(GeoLocation(
ip=ipaddress.ip
) )
)
if geolocations:
response.headers['Cache-Control'] = 'private, max-age=604800'
return geolocations
@app.get("/{ipaddress}")
async def root_get(ipaddress: Annotated[
Union[IPv4Address, IPv6Address],
Path(title="The IPAddress to geolocate")],
response: Response
) -> GeoLocation:
'''
Look up geolocation for ip in path parameter
'''
locations = await root_post([IPAddressParam(ip=ipaddress)], response)
if locations:
response.headers['Cache-Control'] = 'private, max-age=604800'
return locations.pop()
response.status_code = status.HTTP_404_NOT_FOUND
return GeoLocation()
@app.get("/")
def root_redirect(req: Request) -> RedirectResponse:
'''
Redirect empty request using REMOTE_ADDR
'''
return RedirectResponse(url=str(req.url) + str(req.client.host))

View File

@ -1,2 +0,0 @@
geoip2==4.7.0
fastapi==0.115.6

View File

View File

@ -1,24 +0,0 @@
#!/bin/sh
NAME=ismijnipverweg
DIR=/opt/apps/ismijnipverweg/app
USER=www
GROUP=nobody
WORKERS=3
WORKER_CLASS=uvicorn.workers.UvicornWorker
#VENV=$DIR/.venv/bin/activate
BIND=unix:$DIR/../run/gunicorn.sock
LOG_LEVEL=error
cd $DIR
# source $VENV
exec /usr/local/bin/gunicorn main:app \
--name $NAME \
--workers $WORKERS \
--worker-class $WORKER_CLASS \
--user=$USER \
--group=$GROUP \
--bind=$BIND \
--log-level=$LOG_LEVEL \
--log-file=-