Enhance Telegram bot with database integration and API features
- Added SQLite database support with Alembic for migrations. - Implemented FastAPI for HTTP API to manage duties. - Updated configuration to include database URL and HTTP port. - Created entrypoint script for Docker to handle migrations and permissions. - Expanded command handlers to register users and display duties. - Developed a web application for calendar display of duties. - Included necessary Pydantic schemas and SQLAlchemy models for data handling. - Updated requirements.txt to include new dependencies for FastAPI and SQLAlchemy.
This commit is contained in:
47
api/app.py
Normal file
47
api/app.py
Normal file
@@ -0,0 +1,47 @@
|
||||
"""FastAPI app: /api/duties and static webapp."""
|
||||
from pathlib import Path
|
||||
|
||||
import config
|
||||
from fastapi import FastAPI, Query
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
|
||||
from db.session import get_session
|
||||
from db.repository import get_duties
|
||||
from db.schemas import DutyWithUser
|
||||
|
||||
app = FastAPI(title="Duty Teller API")
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=["*"],
|
||||
allow_credentials=True,
|
||||
allow_methods=["*"],
|
||||
allow_headers=["*"],
|
||||
)
|
||||
|
||||
|
||||
@app.get("/api/duties", response_model=list[DutyWithUser])
|
||||
def list_duties(
|
||||
from_date: str = Query(..., description="ISO date YYYY-MM-DD"),
|
||||
to_date: str = Query(..., description="ISO date YYYY-MM-DD"),
|
||||
) -> list[DutyWithUser]:
|
||||
session = get_session(config.DATABASE_URL)
|
||||
try:
|
||||
rows = get_duties(session, from_date=from_date, to_date=to_date)
|
||||
return [
|
||||
DutyWithUser(
|
||||
id=duty.id,
|
||||
user_id=duty.user_id,
|
||||
start_at=duty.start_at,
|
||||
end_at=duty.end_at,
|
||||
full_name=full_name,
|
||||
)
|
||||
for duty, full_name in rows
|
||||
]
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
|
||||
webapp_path = Path(__file__).resolve().parent.parent / "webapp"
|
||||
if webapp_path.is_dir():
|
||||
app.mount("/app", StaticFiles(directory=str(webapp_path), html=True), name="webapp")
|
||||
Reference in New Issue
Block a user