Download and install python: https://www.python.org/downloads/
Then open a powershell window and install flask
pip install flaskIf you are interested in the install location, you can see it with
pip show flaskThen install waitress as professional web service
pip install waitressCreate a folder somewhere and create a file with name app.py
from waitress import serve
from flask import Flask
app = Flask(__name__)
@app.route('/')
def hello():
return "Hello world from Waitress behind IIS!"
if __name__ == '__main__':
serve(app, host='127.0.0.1', port=5000, ident='MyApp', threads=8)If you now execute the file in PowerShell with
python app.pyYou should be able to open the webpage in a local browser with http://localhost:5000
Next step is to make it available via IIS reverse proxy. Create a folder in iis like C:\inetpub\wwwroot\webhook and use URL rewriting on that created subfolder
Just add a reverse proxy rule:
Install ARR, if not already done and do a iisreset:
Then you should be able to access the python app with your domain and subfolder name like https://www.example.com/webhook
To turn the python app off, just press Ctrl + c twice in the powershell window.
If everything was successful, we can create a new python receiver with more advanced functionality
Create a project folder somewhere and create a file like receiver.py
from waitress import serve
from flask import Flask, request, jsonify
import uuid
import os
app = Flask(__name__)
@app.route('/', methods=['POST'])
def webhook():
if request.method == 'POST':
data = request.json
print(f"Received webhook data: {data}")
process_data(request.data) # Process the webhook data here
return jsonify({"message": "Webhook received!"}), 200
else:
return jsonify({"message": "Only POST requests are accepted!"}), 400
def process_data(data):
# process data here
random_file_name = f"{uuid.uuid4()}.txt"
random_file_path = os.path.join(os.getcwd(), random_file_name)
with open(random_file_path, "wb") as file:
file.write(data)
#pass
if __name__ == '__main__':
serve(app, host='127.0.0.1', port=5000, ident='MyApp', threads=8)Start the script with python receiver.py from PowerShell
This script creates a local file with every POST request and puts the payload in it
Now you can call it locally with PowerShell like
Invoke-WebRequest -Method POST -Uri "http://localhost:5000" -ContentType "application/json" -Body '{"event":"order_created","order_id":12345,"customer":"John Doe"}'or via your reverse proxy
Invoke-WebRequest -Method POST -Uri "https://www.example.com/webhook" -ContentType "application/json" -Body '{"event":"order_created","order_id":12345,"customer":"John Doe"}'For a better integration it could be useful to create a windows service. To do that you need to install pywin32
pip install pywin32Then create a service installation script waitress_service.py:
import win32serviceutil
import win32service
import win32event
import servicemanager
from waitress import serve
from receiver import app
class WaitressService(win32serviceutil.ServiceFramework):
_svc_name_ = "WaitressService"
_svc_display_name_ = "Python Waitress Service"
def __init__(self, args):
win32serviceutil.ServiceFramework.__init__(self, args)
self.hWaitStop = win32event.CreateEvent(None, 0, 0, None)
def SvcStop(self):
self.ReportServiceStatus(win32service.SERVICE_STOP_PENDING)
win32event.SetEvent(self.hWaitStop)
def SvcDoRun(self):
servicemanager.LogMsg(servicemanager.EVENTLOG_INFORMATION_TYPE,
servicemanager.PYS_SERVICE_STARTED,
(self._svc_name_, ''))
serve(app, host='127.0.0.1', port=5000, ident='MyApp', threads=8)
if __name__ == '__main__':
win32serviceutil.HandleCommandLine(WaitressService)Just for explanation: The important part is that this script is looking for receiver.py because of from receiver and imports the defined app
Install it with:
python waitress_service.py install
Then you can start your newly created windows service.
The output of this script will by default land in the directory of the service like C:\Users\Administrator\AppData\Local\Programs\Python\Python313
Then you can make changes in the script like defining an absolute path to write the payload
from waitress import serve
from flask import Flask, request, jsonify
import uuid
import os
app = Flask(__name__)
@app.route('/', methods=['POST'])
def webhook():
if request.method == 'POST':
data = request.json
print(f"Received webhook data: {data}")
process_data(request.data) # Process the webhook data here
return jsonify({"message": "Webhook received!"}), 200
else:
return jsonify({"message": "Only POST requests are accepted!"}), 400
def process_data(data):
# process data here
random_file_name = f"{uuid.uuid4()}.txt"
random_file_path = os.path.join("D:\\Scripts\\python\\flask\\webhooks", random_file_name) # use os.getcwd() as current directoy when executing it locally
with open(random_file_path, "wb") as file:
file.write(data)
#pass
if __name__ == '__main__':
serve(app, host='127.0.0.1', port=5000, ident='MyApp', threads=8)After saving it and restart the service, the changes should be reflected immediatly. When the python script cannot be shut down, you can just kill the pythonservice.exe process.
Ensure that the ODBC Driver for SQL Server is installed on your machine. You can download it from the Microsoft website if it's not already installed.
Install pyodbc with
pip install pyodbcCreate a database table in SQLServer like this
USE [Live_Webhooks]
GO
/****** Object: Table [dbo].[Incoming] Script Date: 10.04.2025 18:12:59 ******/
SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
CREATE TABLE [dbo].[Incoming](
[id] [int] IDENTITY(1,1) NOT NULL,
[service] [varchar](50) NULL,
[webhook] [varchar](50) NULL,
[object] [varchar](50) NULL,
[object_id] [varchar](50) NULL,
[event] [varchar](50) NULL,
[payload] [varchar](max) NULL,
[created_at] [datetime] NOT NULL,
CONSTRAINT [PK_Incoming] PRIMARY KEY CLUSTERED
(
[id] ASC
)WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON, OPTIMIZE_FOR_SEQUENTIAL_KEY = OFF) ON [PRIMARY]
) ON [PRIMARY] TEXTIMAGE_ON [PRIMARY]
GO
ALTER TABLE [dbo].[Incoming] ADD CONSTRAINT [DF_Incoming_created_at] DEFAULT (getdate()) FOR [created_at]
GOAnd a user
USE [master]
GO
/* For security reasons the login is created disabled and with a random password. */
CREATE LOGIN [webhooks_service] WITH PASSWORD=N'xT6lWzmqxDPFmVGNVlOKbhHK61h9q+ff3u0W7AH0Jzg=', DEFAULT_DATABASE=[Live_Webhooks], DEFAULT_LANGUAGE=[us_english], CHECK_EXPIRATION=OFF, CHECK_POLICY=OFF
GO
ALTER LOGIN [webhooks_service] DISABLE
GO
USE [Live_Webhooks]
GO
CREATE USER [webhooks_service] FOR LOGIN [webhooks_service] WITH DEFAULT_SCHEMA=[dbo]
GOThat user needs to be activated first.
Then change the script to something like this
[ ] create the rest
This is mostly created by DeepSeek
pip install sqlalchemy, typing, uvicorn, fastapi, aioodbcimport os
import logging
from datetime import datetime
from fastapi import FastAPI, Request, HTTPException, Depends, Header
from fastapi.responses import JSONResponse
from contextlib import asynccontextmanager
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column, Integer, String, Text, DateTime, MetaData
from sqlalchemy.ext.declarative import declarative_base
import uvicorn
import urllib.parse
from typing import Optional
# ----------------------------
# CONFIGURATION
# ----------------------------
# TODO Change current directoy for relative files
os.chdir("D:/Scripts/python/flask/webhooks")
STATIC_TOKEN = os.getenv('STATIC_TOKEN', 'your-static-secret-token-here')
# TODO MSSQL Konfiguration
server = os.getenv('MSSQL_SERVER', 'localhost')
database = os.getenv('MSSQL_DATABASE', 'Live_Webhooks')
username = os.getenv('MSSQL_USERNAME', 'webhooks_service')
password = os.getenv('MSSQL_PASSWORD', 'verysecretpassword')
driver = os.getenv('MSSQL_DRIVER', 'ODBC Driver 17 for SQL Server')
password_encoded = urllib.parse.quote_plus(password)
DATABASE_URL = f"mssql+aioodbc://{username}:{password_encoded}@{server}/{database}?driver={driver}"
# ----------------------------
# CONFIGURE LOGGING
# ----------------------------
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('fastapi_mssql.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
# ----------------------------
# DEFINE LIFESPAN EVENTS
# ----------------------------
@asynccontextmanager
async def lifespan(app: FastAPI):
# Everything until yield gets executed at the start, everything after that at the end
await reflect_tables()
logger.info("Datenbankmetadaten wurden geladen")
logger.info(f"Verfügbare Tabellen: {list(metadata.tables.keys())}")
yield
logger.info("Closing database connection")
sess = get_db()
sess.close()
# ----------------------------
# FASTAPI APP
# ----------------------------
app = FastAPI(
title="Apteco Webhooks Live Event Receiver",
description="Receive events and store them into SQLServer",
version="0.0.1",
lifespan=lifespan
)
# ----------------------------
# CONNECT TO DATABASE AND READ METADATA MODEL
# ----------------------------
engine = create_async_engine(
DATABASE_URL,
pool_size=5,
max_overflow=10,
pool_recycle=300,
pool_pre_ping=True
)
# Metadaten aus bestehender Datenbank laden
metadata = MetaData()
async def reflect_tables():
async with engine.begin() as conn:
await conn.run_sync(metadata.reflect)
# Automatische Tabellenreflektion beim Start
# @app.on_event("startup")
# async def startup():
# await reflect_tables()
# logger.info("Datenbankmetadaten wurden geladen")
# logger.info(f"Verfügbare Tabellen: {list(metadata.tables.keys())}")
AsyncSessionLocal = sessionmaker(
bind=engine,
class_=AsyncSession,
expire_on_commit=False
)
# ----------------------------
# MORE DEPENDENCIES AND FUNCTIONS
# ----------------------------
async def get_db():
async with AsyncSessionLocal() as session:
yield session
# TODO add another function to check a token in the url
async def verify_token(authorization: str = Header(...)):
if not authorization.startswith("Bearer "):
logger.warning("Ungültiges Authorization Header Format")
raise HTTPException(status_code=401, detail="Ungültiges Token Format")
token = authorization.split(" ")[1]
if token != STATIC_TOKEN:
logger.warning(f"Ungültiger Token verwendet: {token}")
raise HTTPException(status_code=401, detail="Ungültiger Token")
return token
# ----------------------------
# ROUTES
# ----------------------------
# TODO CHECK: This is an endpoint
@app.post("/payload")
async def create_payload(
request: Request,
db: AsyncSession = Depends(get_db),
token: str = Depends(verify_token) # TODO CHECK: This is checking the token, could also be a different function
):
try:
data = await request.json()
logger.info(f"Neue Anfrage mit Daten: {str(data)[:200]}...")
if not data:
logger.warning("Leere Anfrage erhalten")
return JSONResponse(
content={"status": "error", "message": "Keine Daten empfangen"},
status_code=400
)
# Zugriff auf die bestehende Tabelle
payloads_table = metadata.tables['Incoming']
# Insert-Operation
# TODO you need to reflect the data that needs to be filled
insert_stmt = payloads_table.insert().values(
service='test',
object='testobject',
object_id=data['order_id'],
event='new',
webhook=str(request.url.path),
payload=str(data)
#timestamp=datetime.utcnow()
)
await db.execute(insert_stmt)
await db.commit()
logger.info("Daten erfolgreich in bestehende Tabelle geschrieben")
return JSONResponse({
"status": "success",
"message": "Daten in MSSQL gespeichert"
})
except KeyError:
logger.error("Tabelle 'payloads' nicht gefunden")
return JSONResponse(
content={"status": "error", "message": "Zieltabelle nicht gefunden"},
status_code=500
)
except Exception as e:
await db.rollback()
logger.error(f"Fehler bei Anfrageverarbeitung: {str(e)}", exc_info=True)
return JSONResponse(
content={"status": "error", "message": str(e)},
status_code=500
)
# ----------------------------
# MAIN
# ----------------------------
if __name__ == "__main__":
logger.info("FastAPI Server wird gestartet...")
uvicorn.run(
app,
host="0.0.0.0",
port=5000,
log_config=None
)This script could also be started as a windows service like in the examples above.
- implement a global token and one per entry?
python receiver_fast.pyTo test it from the same server, you could execute a call via PowerShell like
invoke-webrequest -UseBasicParsing -Uri "http://localhost:5000/payload" -ContentType "application/json" -Method POST -Body '{"event":"order_created","order_id":12345,"customer":"John Doe"}'which should trigger an error because the token is missing.
This one should be working then
invoke-webrequest -UseBasicParsing -Uri "http://localhost:5000/payload" -ContentType "application/json; charset=utf-8 " -Method POST -Body '{"event":"order_created","order_id":123456,"customer":"John Döe"}' -Headers @{"Authorization"="Bearer your-static-secret-token-here"}or like this from external (don't forget the https:):
invoke-restmethod -Uri "https://www.example.com/webhook/payload" -ContentType "application/json; charset=utf-8 " -Method POST -Body '{"event":"order_created","order_id":123,"customer":"John De"}' -Headers @{"Authorization"="Bearer your-static-secret-token-here"}- For production environments, use process managers like
winswverwenden - Activate logging for debugging purposes
serve(app, host='127.0.0.1', port=5000, ident='MyApp') - Increase the threads for waitress:
serve(app, host='127.0.0.1', port=5000, threads=8) - Ensure that the firewall allows traffic for port 5000 and localhost