Getting Started With MongoDB and Sanic

1
git clone git@github.com:mongodb-developer/mongodb-with-sanic.git

1
cd mongodb-with-sanic
2
pip install -r requirements.txt

1
export MONGODB_URL="mongodb+srv://<username>:<password>@<url>/<db>?retryWrites=true&w=majority"

1
python app.py

1
curl -X "POST" "http://localhost:8000/" \
2
    -H 'Accept: application/json' \
3
    -H 'Content-Type: application/json; charset=utf-8' \
4
    -d '{
5
        "name": "Jane Doe",
6
        "email": "jdoe@example.com",
7
        "gpa": "3.9"
8
    }'

1
app = Sanic(__name__)
2
3
settings = dict(
4
    MOTOR_URI=os.environ["MONGODB_URL"],
5
    LOGO=None,
6
)
7
app.config.update(settings)
8
9
BaseModel.init_app(app)
10
11
12
class Student(BaseModel):
13
    __coll__ = "students"

1
@app.route("/", methods=["POST"])
2
async def create_student(request):
3
    student = request.json
4
    student["_id"] = str(ObjectId())
5
6
    new_student = await Student.insert_one(student)
7
    created_student = await Student.find_one(
8
        {"_id": new_student.inserted_id}, as_raw=True
9
    )
10
11
    return json_response(created_student)

1
@app.route("/", methods=["GET"])
2
async def list_students(request):
3
    students = await Student.find(as_raw=True)
4
    return json_response(students.objects)

1
@app.route("/<id>", methods=["GET"])
2
async def show_student(request, id):
3
    if (student := await Student.find_one({"_id": id}, as_raw=True)) is not None:
4
        return json_response(student)
5
6
    raise NotFound(f"Student {id} not found")

1
@app.route("/<id>", methods=["PUT"])
2
async def update_student(request, id):
3
    student = request.json
4
    update_result = await Student.update_one({"_id": id}, {"$set": student})
5
6
    if update_result.modified_count == 1:
7
        if (
8
            updated_student := await Student.find_one({"_id": id}, as_raw=True)
9
        ) is not None:
10
            return json_response(updated_student)
11
12
    if (
13
        existing_student := await Student.find_one({"_id": id}, as_raw=True)
14
    ) is not None:
15
        return json_response(existing_student)
16
17
    raise NotFound(f"Student {id} not found")

1
@app.route("/<id>", methods=["DELETE"])
2
async def delete_student(request, id):
3
    delete_result = await Student.delete_one({"_id": id})
4
5
    if delete_result.deleted_count == 1:
6
        return json_response({}, status=204)
7
8
    raise NotFound(f"Student {id} not found")