b3nw
650ea2d087
All checks were successful
Build and Push Docker Image / build (push) Successful in 34s
75 lines
2.7 KiB
Python
75 lines
2.7 KiB
Python
from fastapi import FastAPI, HTTPException
|
|
import asyncio
|
|
from schwab_scraper import unified_api
|
|
from schwab_scraper.core import Envelope
|
|
|
|
app = FastAPI(title="Schwab Scraper API", version="0.1.0", description="REST API for Schwab Scraper via unified_api")
|
|
browser_lock = asyncio.Semaphore(1)
|
|
|
|
async def check_success(envelope: Envelope):
|
|
if not envelope.get("success"):
|
|
raise HTTPException(status_code=400, detail=envelope.get("error", "Unknown error"))
|
|
return envelope.get("data")
|
|
|
|
@app.get("/api/accounts", tags=["Accounts"])
|
|
async def list_accounts():
|
|
"""List all available Schwab accounts."""
|
|
async with browser_lock:
|
|
env = await unified_api.list_accounts()
|
|
return await check_success(env)
|
|
|
|
@app.get("/api/accounts/overview", tags=["Accounts"])
|
|
async def get_overview(account: str | None = None):
|
|
"""Get a high level overview of an account or all accounts."""
|
|
async with browser_lock:
|
|
env = await unified_api.get_account_overview(account)
|
|
return await check_success(env)
|
|
|
|
@app.get("/api/accounts/positions", tags=["Accounts"])
|
|
async def get_positions(account: str | None = None, include_non_equity: bool = False):
|
|
"""Retrieve positions/holdings for an account."""
|
|
async with browser_lock:
|
|
env = await unified_api.get_positions(account, include_non_equity=include_non_equity)
|
|
return await check_success(env)
|
|
|
|
@app.get("/api/transactions", tags=["Transactions"])
|
|
async def get_transactions(
|
|
account: str | None = None,
|
|
limit: int = 50,
|
|
days_back: int = 90
|
|
):
|
|
"""Fetch transaction history."""
|
|
async with browser_lock:
|
|
env = await unified_api.get_transaction_history_enhanced(
|
|
account=account, limit=limit, days_back=days_back
|
|
)
|
|
return await check_success(env)
|
|
|
|
@app.get("/api/equity/morningstar/{ticker}", tags=["Research"])
|
|
async def get_morningstar(ticker: str):
|
|
"""Get Morningstar rating details for an equity."""
|
|
async with browser_lock:
|
|
env = await unified_api.get_morningstar_data(ticker)
|
|
return await check_success(env)
|
|
|
|
@app.get("/api/equity/phase1/{ticker}", tags=["Research"])
|
|
async def get_equity_phase1(ticker: str):
|
|
"""Fetch base Phase1 equity statistics (pricing, basic facts)."""
|
|
async with browser_lock:
|
|
env = await unified_api.get_equity_phase1_data(ticker)
|
|
return await check_success(env)
|
|
|
|
@app.get("/api/session/status", tags=["System"])
|
|
async def get_session_status():
|
|
"""Check if the cookies and session are currently valid."""
|
|
async with browser_lock:
|
|
env = await unified_api.get_session_status()
|
|
return await check_success(env)
|
|
|
|
def start():
|
|
import uvicorn
|
|
uvicorn.run("schwab_scraper.server.api:app", host="0.0.0.0", port=8000, reload=True)
|
|
|
|
if __name__ == "__main__":
|
|
start()
|