mirror of
https://github.com/LukeHagar/connexion.git
synced 2025-12-06 20:37:47 +00:00
073f0d446e
This PR updates the examples for Connexion 3.0 and merges them for OpenAPI and Swagger. 2 examples required some changes to make them work: - The reverse proxy example required some fixes to the SwaggerUIMiddleware to leverage the `root_path` correctly. This is included in the PR. - The enforced defaults example requires the json validator to adapt the body and pass it on. We currently pass on the original body after validation, and I'm not sure if we should change this. I'll submit a separate PR to discuss this.
64 lines
1.6 KiB
Python
Executable File
64 lines
1.6 KiB
Python
Executable File
import datetime
|
|
import logging
|
|
|
|
import connexion
|
|
import orm
|
|
from connexion import NoContent
|
|
|
|
db_session = None
|
|
|
|
|
|
def get_pets(limit, animal_type=None):
|
|
q = db_session.query(orm.Pet)
|
|
if animal_type:
|
|
q = q.filter(orm.Pet.animal_type == animal_type)
|
|
return [p.dump() for p in q][:limit]
|
|
|
|
|
|
def get_pet(pet_id):
|
|
pet = db_session.query(orm.Pet).filter(orm.Pet.id == pet_id).one_or_none()
|
|
return pet.dump() if pet is not None else ("Not found", 404)
|
|
|
|
|
|
def put_pet(pet_id, pet):
|
|
p = db_session.query(orm.Pet).filter(orm.Pet.id == pet_id).one_or_none()
|
|
pet["id"] = pet_id
|
|
if p is not None:
|
|
logging.info("Updating pet %s..", pet_id)
|
|
p.update(**pet)
|
|
else:
|
|
logging.info("Creating pet %s..", pet_id)
|
|
pet["created"] = datetime.datetime.utcnow()
|
|
db_session.add(orm.Pet(**pet))
|
|
db_session.commit()
|
|
return NoContent, (200 if p is not None else 201)
|
|
|
|
|
|
def delete_pet(pet_id):
|
|
pet = db_session.query(orm.Pet).filter(orm.Pet.id == pet_id).one_or_none()
|
|
if pet is not None:
|
|
logging.info("Deleting pet %s..", pet_id)
|
|
db_session.query(orm.Pet).filter(orm.Pet.id == pet_id).delete()
|
|
db_session.commit()
|
|
return NoContent, 204
|
|
else:
|
|
return NoContent, 404
|
|
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
db_session = orm.init_db("sqlite:///:memory:")
|
|
app = connexion.FlaskApp(__name__, specification_dir="spec")
|
|
app.add_api("openapi.yaml")
|
|
app.add_api("swagger.yaml")
|
|
|
|
application = app.app
|
|
|
|
|
|
@application.teardown_appcontext
|
|
def shutdown_session(exception=None):
|
|
db_session.remove()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(port=8080, reload=False)
|