[Python] FastAPI and Couchbase

Discussão em 'Python' iniciado por Stack, Dezembro 22, 2024.

  1. Stack

    Stack Membro Participativo

    I’m currently working on a CRUD application using FastAPI and Couchbase, but I’ve encountered a problem with the GET method—it doesn’t seem to be functioning as expected. I’ve tried debugging but haven’t been able to identify the issue. Could someone provide guidance or help me resolve this?

    Here is the code:

    from couchbase.cluster import Cluster, ClusterOptions
    from couchbase.auth import PasswordAuthenticator
    from couchbase.exceptions import DocumentNotFoundException
    from couchbase.bucket import Bucket

    class Error(BaseModel):
    description: str
    error_code: str
    error_id: int
    timestamp: str

    class Location(BaseModel):
    coordinates: str
    location_id: int
    room_name: str

    class MaintenanceHistory(BaseModel):
    description: str
    maintenance_date: str
    maintenance_id: int
    parts: list

    class User(BaseModel):
    address: str
    email: str
    name: str
    phone_number: str
    user_id: int

    class Task(BaseModel):
    scheduled_time: str
    status: str
    task_id: int
    task_type: str
    user: User

    class Robot(BaseModel):
    robot_id: int
    model: str
    serial_number: str
    status: str
    errors: list[Error]
    locations: list[Location]
    maintenance_history: list[MaintenanceHistory]
    tasks: list[Task]
    usage_history: list
    user_settings: list

    # FastAPI app and CORS middleware setup
    app = FastAPI()

    origins = [
    "http://127.0.0.1:5500",
    "http://localhost:5500",
    ]

    app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    )

    # Couchbase connection setup (same as before)
    cluster = Cluster('couchbases://cb.flgdpdewcbi5.cloud.couchbase.com', ClusterOptions(PasswordAuthenticator('Robot', ‘dddddd’)))
    bucket = cluster.bucket('robot_system')
    collection = bucket.scope('system').collection('Robot')

    # Endpoints (same as before)
    @app.get("/robots/{robot_id}", response_model=Robot)
    async def get_robot(robot_id: int):
    try:
    document = collection.get(str(robot_id))
    return document.content_as[Robot]
    except DocumentNotFoundException:
    raise HTTPException(status_code=404, detail="Robot not found")```

    Continue reading...

Compartilhe esta Página