Compare commits
1 Commits
main
...
b100d24273
Author | SHA1 | Date | |
---|---|---|---|
b100d24273
|
@ -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')
|
|
@ -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')
|
|
@ -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')
|
|
89
app/main.py
89
app/main.py
@ -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))
|
|
||||||
|
@ -1,2 +0,0 @@
|
|||||||
geoip2==4.7.0
|
|
||||||
fastapi==0.115.6
|
|
24
start.sh
24
start.sh
@ -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=-
|
|
Reference in New Issue
Block a user