minte9

LearnRemember   PYTHON

FastAPI

==== APIS vs SHELL ==== Centralizing these actions into an API is a much cleaner and scalable approach than invoking PowerShell scripts from PHP: Decoupling: PHP pages won’t need to know about PowerShell or Python internals. Scalability: Easily add new actions without modifying PHP code. Security: Avoid shell execution vulnerabilities. Maintainability: Centralized logic in one service. Logging: Allow better error handling and logging Async: Enable asynchronous execution if needed ==== Recommended Architecture ==== Create a Python-based API service (using FastAPI or Flask). Expose endpoints for each action (e.g., /hello). Run Python scripts directly inside the API (no need for PowerShell). PHP calls the API via HTTP instead of shell_exec. ==== FastAPI service with one endpoint ====

Create Project Structure

 
D:\python-apps\fastapi-service\

│
├── app\
│   ├── main.py
│   ├── endpoints\
│   │   ├── hello_world.py
│
├── requirements.txt

Install Dependencies

 
virtualenv ./myenv
pip install fastapi uvicorn
pip freeze > requirements.txt

FastAPI Code

app/main.py

 
from fastapi import FastAPI
from app.endpoints import hello_world

app = FastAPI(title="My Automations API")

# Register hello_world router
app.include_router(hello_world.router, prefix="/hello", tags=["Hello World"])

app/endpoints/hello_world.py

 
from fastapi import APIRouter

router = APIRouter()

@router.get("/")
async def hello():
    return {"message": "Hello from FastAPI!"}

Run the API

From D:\python-apps\fastapi-service\:

 
uvicorn app.main:app --host 0.0.0.0 --port 8000
        
python -B -m uvicorn app.main:app --host 0.0.0.0 --port 8000
    # B tells Python not to write __pycache__

Now your API is live at: http://localhost:8000/hello/ Test in browser or with curl: curl http://localhost:8000/hello/ Expected response: {"message": "Hello from FastAPI!"}

PHP Controller (Async Request):

Snippet using file_get_contents (or you can use curl for more control):

 
if (!empty($_POST['btn_process'])) {
    $url = "http://localhost:8000/hello/";
    $response = file_get_contents($url);
    $data['message'] = $response;
}

Enable HTTPS in FastAPI (Uvicorn)

You need SSL certificate private key in .env

 
HOST=0.0.0.0
PORT=8443
SSL_CRT=D:\certs\server.crt
SSL_KEY=D:\certs\server.key

Command to run FastAPI with SSL:

 
python -B -m uvicorn app.main:app --host 0.0.0.0 --port 8443 --ssl-keyfile "%SSL_KEY%" --ssl-certfile "%SSL_CRT%"

If not found error (just for testing, add paths):

 
python -B -m uvicorn app.main:app --host 0.0.0.0 --port 8443 --ssl-keyfile "D:\certs\server.key"  --ssl-certfile "D:\certs\server.crt"

Programmatic start with uvicorn.run().

Create a small run.py that: Loads environment variables (from .env or system) Builds an SSLContext Starts Uvicorn programmatically app/run.py

 
"""
    python -B run_server.py
"""

import os
import ssl
import uvicorn

from dotenv import load_dotenv
load_dotenv()

HOST = os.getenv("HOST", "0.0.0.0")
PORT = int(os.getenv("PORT", "8443"))

SSL_CERT = os.getenv("SSL_CRT")
SSL_KEY  = os.getenv("SSL_KEY")

if __name__ == "__main__":
    uvicorn.run("app.main:app", host=HOST, port=PORT, 
    ssl_certfile=SSL_CERT,
    ssl_keyfile=SSL_KEY
)

Unit Testing

Install dependencies.

 
.\myenv\Scripts\Activate
(myenv) pip install pytest requests httpx
(myenv) pip freeze > requirements.txt

Test endpoint (no server needed).

 
""" Run test in quite mode (or with no bytecode cache):

    cd fastapi-service/
    ./myenv/Scripts/activate
    (myenv) pip install pytest
    (myenv) pytest -q
    (myenv) python -B -m pytest
"""

import sys
sys.dont_write_bytecode = True  # no .pyc

from fastapi.testclient import TestClient
from app.main import app  # Import your FastApi app

def test_hello_endpoint():
    client = TestClient(app)
    response = client.get("/hello/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello from FastAPI"}

If running from the root still gives issues, add a pytest.ini at the project root:

 
# D:\python-apps\fastapi-service\pytest.ini
[pytest]
testpaths = tests
pythonpath = .

Run test in quite mode (or with no bytecode cache)

 
(myenv) pytest -q
(myenv) python -B -m pytest

Test the live UAT URL (like a PHP snippet)

 
""" Run test in quite mode (or with no bytecode cache):
    tests/test_hello.py
    
    cd fastapi-service/
    ./myenv/Scripts/activate
    (myenv) pip install pytest requests httpx
    (myenv) python -B -m pytest -q
    (myenv) python -B -m pytest -q -m integration
"""

import os
import requests
import pytest
from fastapi.testclient import TestClient
from app.main import app
from dotenv import load_dotenv
load_dotenv() 

@pytest.mark.unit
def test_hello_endpoint():
    client = TestClient(app)
    response = client.get("/fastapi/hello_world/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello from FastAPI - GET"}

@pytest.mark.integration
def test_hello_endpoint_real_server():
    SERVER_URL = os.getenv('SERVER_URL')
    SSL_CRT = os.getenv('SSL_CRT')
    SSL_KEY = os.getenv('SSL_KEY')
    response = requests.post(SERVER_URL + '/fastapi/hello_world/second', timeout=8, verify=SSL_CRT)  # verify=False
    data = response.json()
    assert response.status_code == 200
    assert data['message'] == 'Hello from FastAPI - POST'

Logging

Using Uvicorn's default logging.

 
# server.py
import uvicorn

if __name__ == "__main__":
    # This uses Uvicorn's built-in logging configuration
    uvicorn.run(
        "app:app",           # module:variable
        host="127.0.0.1",
        port=8000,
        log_level="info",    # info/debug/warning/error/critical
    )

app.py

 

from fastapi import FastAPI
import logging

app = FastAPI()

# Use a module-level logger
logger = logging.getLogger(__name__)

@app.get("/hello")
def hello():
    logger.info("Handling /hello")   # This will appear in the server logs

Run server

 
python server.py

For tests (pytest), logs are often captured. To show them on the test terminal, add a pytest.ini:

 

[pytest]
testpaths = tests
pythonpath = .
log_cli = false
log_cli_level = INFO
log_cli_format = %(asctime)s - %(levelname)s - %(message)s
log_cli_date_format = %Y-%m-%d %H:%M:
markers = 
    integration: tests that require real network/services (slower)
    unit: testclient (fast)

Service wrapper

Install pywin32 globally.

 
PS D:\Program Files\Python39> .\python.exe -m pip install pywin32
PS D:\Program Files\Python39> .\python.exe .\Scripts\pywin32_postinstall.py -install

Stop any other pywin32 service, if it' running (important).

 
D:\python-apps\fastapi-service>python fastapi_service.py install
D:\python-apps\fastapi-service>python fastapi_service.py start

Service wrapper.

 
""" FASTAPI SERVICE - WRAPPER

Windows PowerShell:
cd D:\python-apps\fastapi-service
.\myenv\Scripts\activate
(myenv) pip install pywin32
(myenv) D:\python-apps\fastapi-service>

Install:
python fastapi_service.py install
python fastapi_service.py start

Reinstall:
python fastapi_service.py stop
python fastapi_service.py remove
"""

import win32serviceutil
import win32service
import win32event
import servicemanager
import subprocess
import sys
import os

class FastAPIService(win32serviceutil.ServiceFramework):
    _svc_name_ = "AppPPF_FastAPIService"
    _svc_display_name_ = "AppPPF FastAPI v1.0"
    _svc_description_ = "PPF FastAPI Application Service"

    def __init__(self, args):
        win32serviceutil.ServiceFramework.__init__(self, args)
        self.hWaitStop = win32event.CreateEvent(None, 0, 0, None)
        self.process = None

    def SvcStop(self):
        self.ReportServiceStatus(win32service.SERVICE_STOP_PENDING)
        win32event.SetEvent(self.hWaitStop)
        if self.process:
            self.process.terminate()

    def SvcDoRun(self):
        servicemanager.LogInfoMsg("Starting FastAPI...")
        # Path to your virtual environment Python
        venv_python = "D:/python-apps/fastapi-service/venv/Scripts/python.exe"
        # Path to your FastAPI app
        app_script = "D:/python-apps/fastapi-service/server.py"
        # Start FastAPI app
        self.process = subprocess.Popen([venv_python, "-B", app_script])
        win32event.WaitForSingleObject(self.hWaitStop, win32event.INFINITE)

if __name__ == '__main__':
    win32serviceutil.HandleCommandLine(FastAPIService)

Migrations

 
(myenv) pip freeze > requirements.txt
(myenv) pip install -r requirements.txt

Full Example (app.main)

 
""" FastAPI app definition & router includes
"""

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from dotenv import load_dotenv
import logging
import os

# * Load .evn (explicit path for Windows Service)
load_dotenv(r"D:\fastapi-service\.env")

# * Logging configuration
logging.basicConfig(
    level=os.getenv("LOG_LEVEL", "INFO").upper(),
    filename=os.getenv("LOG_FILE", "d:/fastapi-service/logs/service.log"),
    filemode="a",
    format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
    datefmt="%Y-%m-%h %H:%M:%S"
)

app = FastAPI(title="PPF Automations API", root_path="/fastapi")
logger = logging.getLogger(__name__)

# Routers
from app.endpoints import router_hello_world, router_import_mtm_options
app.include_router(router_hello_world.router, prefix="/hello_world", tags=["Hello World"])
app.include_router(router_import_my_options.router, prefix="/import_my_options", tags=["File Parsers"])

# Allowed IPs
ALLOWED_IPS = {ip for ip in os.getenv("ALLOWED_IPS").split(",")}

@app.middleware("http")
async def id_denied_middleware(request: Request, call_next):
    client_ip = request.client.host

    if client_ip not in ALLOWED_IPS:
        logger.warning(f"DENY from {client_ip}")
        return JSONResponse({"status":"error", "output": f"Access denied from IP {client_ip}"})

    response = await call_next(request)
    logger.info(f"Response: {response.status_code} for {request.url.path} {client_ip}")
    return response