142 lines
4.3 KiB
Python
142 lines
4.3 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.orm import Session
|
|
import uuid
|
|
from models.routing_api import RoutingAPI
|
|
from schemas.routing_api import RoutingAPICreate, RoutingAPIUpdate, RoutingAPISchema
|
|
|
|
def get_routing_api_by_id(db: Session, routing_api_id: uuid.UUID) -> Optional[RoutingAPI]:
|
|
"""
|
|
Retrieves a routing API by its ID.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
routing_api_id (UUID): The ID of the routing API to retrieve.
|
|
|
|
Returns:
|
|
Optional[RoutingAPI]: The routing API object if found, otherwise None.
|
|
"""
|
|
return db.query(RoutingAPI).filter(RoutingAPI.id == routing_api_id).first()
|
|
|
|
def get_all_routing_apis(db: Session) -> List[RoutingAPI]:
|
|
"""
|
|
Retrieves all routing APIs from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
|
|
Returns:
|
|
List[RoutingAPI]: A list of all routing API objects.
|
|
"""
|
|
return db.query(RoutingAPI).all()
|
|
|
|
def create_routing_api(db: Session, routing_api_data: RoutingAPICreate) -> RoutingAPI:
|
|
"""
|
|
Creates a new routing API in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
routing_api_data (RoutingAPICreate): The data for the routing API to create.
|
|
|
|
Returns:
|
|
RoutingAPI: The newly created routing API object.
|
|
"""
|
|
db_routing_api = RoutingAPI(**routing_api_data.dict())
|
|
db.add(db_routing_api)
|
|
db.commit()
|
|
db.refresh(db_routing_api)
|
|
return db_routing_api
|
|
|
|
def update_routing_api(db: Session, routing_api_id: uuid.UUID, routing_api_data: RoutingAPIUpdate) -> RoutingAPI:
|
|
"""
|
|
Updates an existing routing API in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
routing_api_id (UUID): The ID of the routing API to update.
|
|
routing_api_data (RoutingAPIUpdate): The updated data for the routing API.
|
|
|
|
Returns:
|
|
RoutingAPI: The updated routing API object.
|
|
"""
|
|
db_routing_api = db.query(RoutingAPI).filter(RoutingAPI.id == routing_api_id).first()
|
|
if not db_routing_api:
|
|
raise ValueError(f"Routing API with ID {routing_api_id} not found")
|
|
|
|
update_data = routing_api_data.dict(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(db_routing_api, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_routing_api)
|
|
return db_routing_api
|
|
|
|
def delete_routing_api(db: Session, routing_api_id: uuid.UUID) -> bool:
|
|
"""
|
|
Deletes a routing API from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
routing_api_id (UUID): The ID of the routing API to delete.
|
|
|
|
Returns:
|
|
bool: True if the routing API was deleted, False otherwise.
|
|
"""
|
|
db_routing_api = db.query(RoutingAPI).filter(RoutingAPI.id == routing_api_id).first()
|
|
if not db_routing_api:
|
|
return False
|
|
|
|
db.delete(db_routing_api)
|
|
db.commit()
|
|
return True
|
|
|
|
def generate_api_key(db: Session, routing_api: RoutingAPI) -> str:
|
|
"""
|
|
Generates a unique API key for a routing API.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
routing_api (RoutingAPI): The routing API object.
|
|
|
|
Returns:
|
|
str: The generated API key.
|
|
"""
|
|
existing_keys = db.query(RoutingAPI.api_key).all()
|
|
existing_keys = [key[0] for key in existing_keys]
|
|
|
|
api_key = str(uuid.uuid4())
|
|
while api_key in existing_keys:
|
|
api_key = str(uuid.uuid4())
|
|
|
|
routing_api.api_key = api_key
|
|
db.commit()
|
|
db.refresh(routing_api)
|
|
return api_key
|
|
|
|
def get_routing_api_schema(db_routing_api: RoutingAPI) -> RoutingAPISchema:
|
|
"""
|
|
Converts a RoutingAPI model object to a RoutingAPISchema object.
|
|
|
|
Args:
|
|
db_routing_api (RoutingAPI): The RoutingAPI model object.
|
|
|
|
Returns:
|
|
RoutingAPISchema: The corresponding RoutingAPISchema object.
|
|
"""
|
|
return RoutingAPISchema.from_orm(db_routing_api)
|
|
|
|
def get_routing_api_requests_count(db: Session, routing_api_id: uuid.UUID) -> int:
|
|
"""
|
|
Retrieves the number of requests made to a routing API.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
routing_api_id (UUID): The ID of the routing API.
|
|
|
|
Returns:
|
|
int: The number of requests made to the routing API.
|
|
"""
|
|
routing_api = db.query(RoutingAPI).filter(RoutingAPI.id == routing_api_id).first()
|
|
if not routing_api:
|
|
raise ValueError(f"Routing API with ID {routing_api_id} not found")
|
|
|
|
return len(routing_api.routes) |