Sqlalchemty

Нам понадобится установить пакет psycopg2-binary.

pip install psycopg2-binary

• SQLalchemy библиотека Python с её помощью мы можем выполнять SQL запросы.
• Начнём с кода выводящего информацию из таблицы flights:

  	
  import os
  from sqlalchemy import create_engine
  from sqlalchemy.orm import scoped_session, sessionmaker
  from sqlalchemy.sql import text
  database_url = os.getenv('DATABASE_URL')
  engine = create_engine(database_url)
  db = scoped_session(sessionmaker(bind=engine))
  
  def main():
  	flights = db.execute(text('SELECT origin, destination, duration FROM flights')).fetchall()
  	for flight in flights:
  		print(f"{flight.origin} to {flight.destination}, {flight.duration} minutes.")
  
  if __name__ == "__main__":
  	main()
  	
  	

• Запрос на вставку данных в таблицу из csv файла:

  		
  import csv
  import os
  
  from sqlalchemy import create_engine
  from sqlalchemy.orm import scoped_session, sessionmaker
  from sqlalchemy.sql import text
  
  engine = create_engine(os.getenv("DATABASE_URL"))
  db = scoped_session(sessionmaker(bind=engine))
  
  def main():
      f = open("flights.csv")
      reader = csv.reader(f)
      for origin, destination, duration in reader:
          db.execute(text("INSERT INTO flights (origin, destination, duration) VALUES (:origin, :destination, :duration)"),
                      {"origin": origin, "destination": destination, "duration": duration})
          print(f"Added flight from {origin} to {destination} lasting {duration} minutes.")
      db.commit()
  
  if __name__ == "__main__":
      main()
      
  		

Sqlalchemy с Flask

• Приведём пример кода:

  		
  import os
  from flask import Flask, render_template, request, url_for
  from sqlalchemy import create_engine
  from sqlalchemy.sql import text
  from sqlalchemy.orm import scoped_session, sessionmaker
  database_url = os.getenv("DATABASE_URL")
  engine = create_engine(database_url)
  db = scoped_session(sessionmaker(bind=engine))
  
  app = Flask(__name__)
  
  @app.route('/')
  def index():
      flights = db.execute(text('select * from flights')).fetchall()
      return render_template('index.html', flights = flights)
  
  @app.route('/book', methods=["POST"])
  def book():
      """Book a flight."""
  
      # Get form information.
      name = request.form.get("name")
      try:
          flight_id = int(request.form.get("flight_id"))
      except ValueError:
          return render_template("error.html", message="Invalid flight number.")
  
      # Make sure the flight exists.
      if db.execute(text("SELECT * FROM flights WHERE id = :id"), {"id": flight_id}).rowcount == 0:
          return render_template("error.html", message="No such flight with that id.")
      db.execute(text("INSERT INTO passengers (name, flight_id) VALUES (:name, :flight_id)"),
              {"name": name, "flight_id": flight_id})
      db.commit()
      return render_template("success.html")
  
  @app.route("/flights")
  def flights():
      """Lists all flights."""
      flights = db.execute(text("SELECT * FROM flights")).fetchall()
      return render_template("flights.html", flights=flights)
  
  @app.route("/flights/<int:flight_id>")
  def flight(flight_id):
      """Lists details about a single flight."""
  
      # Make sure flight exists.
      flight = db.execute(text("SELECT * FROM flights WHERE id = :id"), {"id": flight_id}).fetchone()
      if flight is None:
          return render_template("error.html", message="No such flight.")
  
      # Get all passengers.
  
  		

• Создадим flights.html

• Создадим index.html.

• flight.html