Compare commits

...

10 Commits

7 changed files with 107 additions and 11 deletions

View File

@ -0,0 +1,17 @@
---
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')

17
.gitea/workflows/mypy.yml Normal file
View File

@ -0,0 +1,17 @@
---
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

@ -0,0 +1,17 @@
---
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

@ -6,7 +6,8 @@ from typing import Annotated, Optional, Union
import geoip2.database import geoip2.database
from geoip2.errors import AddressNotFoundError from geoip2.errors import AddressNotFoundError
from fastapi import FastAPI, Path, Body, Response, status 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()
@ -14,11 +15,13 @@ 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):
''' '''
@ -29,19 +32,23 @@ 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[list[IPAddressParam], async def root_post(ipaddresses: Annotated[
Body(title="The IPAddresses to geolocate")] list[IPAddressParam],
Body(title="The IPAddresses to geolocate")],
response: Response
) -> list[GeoLocation]: ) -> list[GeoLocation]:
''' '''
Return GeoLocation item(s) for a list of IPAddressParam objects Return GeoLocation item(s) for a list of IPAddressParam objects
@ -71,19 +78,31 @@ async def root_post(ipaddresses: Annotated[list[IPAddressParam],
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[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 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)]) locations = await root_post([IPAddressParam(ip=ipaddress)], response)
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("/")
def root_redirect(req: Request) -> RedirectResponse:
'''
Redirect empty request using REMOTE_ADDR
'''
return RedirectResponse(url=str(req.url) + str(req.client.host))

2
requirements.txt Normal file
View File

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

0
run/.gitkeep Normal file
View File

24
start.sh Executable file
View File

@ -0,0 +1,24 @@
#!/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=-