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') | ||||
							
								
								
									
										109
									
								
								app/main.py
									
									
									
									
									
								
							
							
						
						
									
										109
									
								
								app/main.py
									
									
									
									
									
								
							| @ -1,13 +1,12 @@ | ||||
| ''' | ||||
| Simple Geolocation with FastAPI | ||||
| ''' | ||||
| import dataclasses | ||||
| from ipaddress import IPv4Address, IPv4Network, IPv6Address, IPv6Network | ||||
| from typing import Annotated, Optional, Union | ||||
|  | ||||
| import geoip2.database | ||||
| from geoip2.errors import AddressNotFoundError | ||||
| from fastapi import FastAPI, Path, Body, Request, Response, status | ||||
| from fastapi.responses import RedirectResponse | ||||
| from fastapi import FastAPI, Path | ||||
| from pydantic import BaseModel | ||||
|  | ||||
| app = FastAPI() | ||||
| @ -15,94 +14,46 @@ app = FastAPI() | ||||
| GEOLITE2_ASN_DB = '/usr/local/share/GeoIP/GeoLite2-ASN.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): | ||||
|     ''' | ||||
|     Locality data | ||||
|     ''' | ||||
|     city: Optional[str] = None | ||||
|     country: Optional[str] = None | ||||
|     continent: Optional[str] = None | ||||
|     is_eu: bool = False | ||||
|  | ||||
|     city: Optional[str] | ||||
|     country: Optional[str] | ||||
|     continent: Optional[str] | ||||
|     is_eu: bool | ||||
|  | ||||
| class GeoLocation(BaseModel): | ||||
|     ''' | ||||
|     Geolocation data model | ||||
|     ''' | ||||
|     ip: Optional[Union[IPv6Address, IPv4Address]] = None | ||||
|     asn: Optional[int] = None | ||||
|     asn_org: Optional[str] = None | ||||
|     network: Optional[Union[IPv6Network, IPv4Network]] = None | ||||
|     locality: Locality = Locality() | ||||
|  | ||||
|  | ||||
| @app.post("/") | ||||
| async def root_post(ipaddresses: Annotated[ | ||||
|     list[IPAddressParam], | ||||
|     Body(title="The IPAddresses to geolocate")], | ||||
|                     response: Response | ||||
|                     ) -> list[GeoLocation]: | ||||
|     ''' | ||||
|     Return GeoLocation item(s) for a list of IPAddressParam objects | ||||
|     ''' | ||||
|     geolocations = [] | ||||
|     with (geoip2.database.Reader(GEOLITE2_ASN_DB) as reader_asn, | ||||
|             geoip2.database.Reader(GEOLITE2_CITY_DB) as reader_city): | ||||
|         for ipaddress in ipaddresses: | ||||
|             try: | ||||
|                 asn_data = reader_asn.asn(ipaddress.ip) | ||||
|                 city_data = reader_city.city(ipaddress.ip) | ||||
|  | ||||
|                 geolocations.append(GeoLocation( | ||||
|                     ip=ipaddress.ip, | ||||
|                     asn=asn_data.autonomous_system_number, | ||||
|                     asn_org=asn_data.autonomous_system_organization, | ||||
|                     network=asn_data.network, | ||||
|                     locality=Locality( | ||||
|                         city=city_data.city.name, | ||||
|                         country=city_data.country.iso_code, | ||||
|                         continent=city_data.continent.code, | ||||
|                         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 | ||||
|  | ||||
|     ip: Union[IPv6Address,IPv4Address] | ||||
|     asn: Optional[int] | ||||
|     asn_org: Optional[str] | ||||
|     network: Union[IPv6Network,IPv4Network,None] | ||||
|     locality: Locality | ||||
|  | ||||
| @app.get("/{ipaddress}") | ||||
| async def root_get(ipaddress: Annotated[ | ||||
|     Union[IPv4Address, IPv6Address], | ||||
|     Path(title="The IPAddress to geolocate")], | ||||
|                    response: Response | ||||
|                    ) -> GeoLocation: | ||||
| async def root(ipaddress: Annotated[Union[IPv4Address,IPv6Address], | ||||
|                                     Path(title="The IPAddress to geolocate")] | ||||
|                ) -> 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() | ||||
|     with (geoip2.database.Reader(GEOLITE2_ASN_DB) as reader_asn, | ||||
|             geoip2.database.Reader(GEOLITE2_CITY_DB) as reader_city): | ||||
|         asn_data = reader_asn.asn(ipaddress) | ||||
|         city_data = reader_city.city(ipaddress) | ||||
|  | ||||
|  | ||||
| @app.get("/") | ||||
| def root_redirect(req: Request) -> RedirectResponse: | ||||
|     ''' | ||||
|     Redirect empty request using REMOTE_ADDR | ||||
|     ''' | ||||
|     return RedirectResponse(url=str(req.url) + str(req.client.host)) | ||||
|         return GeoLocation( | ||||
|             ip=ipaddress, | ||||
|             asn=asn_data.autonomous_system_number, | ||||
|             asn_org=asn_data.autonomous_system_organization, | ||||
|             network=asn_data.network, | ||||
|             locality=Locality( | ||||
|                 city=city_data.city.name, | ||||
|                 country=city_data.country.iso_code, | ||||
|                 continent=city_data.continent.code, | ||||
|                 is_eu=city_data.country.is_in_european_union | ||||
|             ) | ||||
|         ) | ||||
|  | ||||
| @ -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