Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# Following method is exposed through the REST API
# This means it can be invoked with a HTTP POST
@classmethod
@jsonapi_rpc(http_methods=["POST"])
def send_mail(self, **args):
"""
description : Send an email
args:
email:
type : string
example : test email
"""
return {"result": args}
class Book(SAFRSBase, db.Model):
"""
description: Book description
"""
__tablename__ = "Books"
id = db.Column(db.String, primary_key=True)
name = db.Column(db.String, default="")
user_id = db.Column(db.String, db.ForeignKey("Users.id"))
user = db.relationship("User", back_populates="books")
def populate_db():
"""
Add some items to the db
"""
user0 = User(name="user0", email="em@il0")
mailfile.write(content)
return {"result": "sent {}".format(content)}
class Publisher(SAFRSBase, db.Model):
"""
description: Publisher description
"""
__tablename__ = "Publishers"
id = db.Column(db.String, primary_key=True)
name = db.Column(db.String, default="")
books = db.relationship("Book", back_populates="publisher")
class Review(SAFRSBase, db.Model):
"""
description: Review description
"""
__tablename__ = "Reviews"
reader_id = db.Column(db.String, db.ForeignKey("People.id"), primary_key=True)
book_id = db.Column(db.String, db.ForeignKey("Books.id"), primary_key=True)
review = db.Column(db.String, default="")
person = db.relationship(Person)
book = db.relationship(Book)
class User(SAFRSBase, db.Model):
__tablename__ = "Users"
id = db.Column(db.String(300), primary_key=True)
username = db.Column(db.String(32), index=True)
import sys, logging, inspect, builtins
from sqlalchemy import CHAR, Column, DateTime, Float, ForeignKey, Index, Integer, String, TIMESTAMP, Table, Text, UniqueConstraint, text
from sqlalchemy.sql.sqltypes import NullType
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from flask_sqlalchemy import SQLAlchemy
from flask import Flask, redirect
from flask_swagger_ui import get_swaggerui_blueprint
from safrs import SAFRSBase, jsonapi_rpc, SAFRSJSONEncoder, Api
from safrs import search, startswith
app = Flask("SAFRS Demo App")
app.config.update(SQLALCHEMY_DATABASE_URI="mysql+pymysql://root:password@localhost/mysql", DEBUG=True)
SAFRSBase.db_commit = False
builtins.db = SQLAlchemy(app) # set global variables to be used in the import
builtins.SAFRSBase = SAFRSBase
import mysql
def start_api(HOST="0.0.0.0", PORT=80):
with app.app_context():
api = Api(app, api_spec_url="/api/swagger", host="{}:{}".format(HOST, PORT), schemes=["http"], description="")
# Get the SAFRSBase models from mysql
for name, model in inspect.getmembers(mysql):
bases = getattr(model, "__bases__", [])
if SAFRSBase in bases:
# Create an API endpoint
api.expose_object(model)
import logging
from flask import Flask, redirect, request
from flask_sqlalchemy import SQLAlchemy
from flask_swagger_ui import get_swaggerui_blueprint
from flask_cors import CORS
from safrs import SAFRSBase, SAFRSAPI, jsonapi_rpc
from safrs.safrs_types import SAFRSID
from safrs.util import classproperty
from collections import namedtuple
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm.interfaces import ONETOMANY, MANYTOMANY # , MANYTOONE
import pdb
db = SQLAlchemy()
class User(SAFRSBase, db.Model):
"""
description: User description
"""
__tablename__ = "Users"
id = db.Column(db.String, primary_key=True)
name = db.Column(db.String, default="")
email = db.Column(db.String, default="")
books = db.relationship("Book", back_populates="user", lazy="dynamic")
class Book(SAFRSBase, db.Model):
"""
description: Book description
"""
db = SQLAlchemy()
auth = HTTPBasicAuth()
# Example sqla database object
class Item(SAFRSBase, db.Model):
"""
description: Item description
"""
__tablename__ = "items"
id = Column(String, primary_key=True)
name = Column(String, default="")
class User(SAFRSBase, db.Model):
"""
description: User description
"""
__tablename__ = "users"
username = db.Column(db.String(32), primary_key=True)
password_hash = db.Column(db.String(64))
custom_decorators = [auth.login_required]
@jsonapi_rpc(http_methods=["POST"])
def hash_password(self, password):
self.password_hash = pwd_context.encrypt(password)
@jsonapi_rpc(http_methods=["POST"])
def verify_password(self, password):
return pwd_context.verify(password, self.password_hash)
t_staff_list = Table(
"staff_list",
metadata,
Column("ID", TINYINT(3), server_default=text("'0'")),
Column("name", String(91)),
Column("address", String(50)),
Column("zip code", String(10)),
Column("phone", String(20)),
Column("city", String(50)),
Column("country", String(50)),
Column("SID", TINYINT(3)),
)
class Store(SAFRSBase, Base):
__tablename__ = "store"
store_id = Column(TINYINT(3), primary_key=True)
manager_staff_id = Column(ForeignKey("staff.staff_id", onupdate="CASCADE"), nullable=False, unique=True)
address_id = Column(ForeignKey("address.address_id", onupdate="CASCADE"), nullable=False, index=True)
last_update = Column(TIMESTAMP, nullable=False, server_default=text("CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"))
address = relationship("Addres")
manager_staff = relationship("Staff", primaryjoin="Store.manager_staff_id == Staff.staff_id")
import sys, logging, inspect, builtins
from sqlalchemy import CHAR, Column, DateTime, Float, ForeignKey, Index, Integer, String, TIMESTAMP, Table, Text, UniqueConstraint, text
from sqlalchemy.sql.sqltypes import NullType
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from flask_sqlalchemy import SQLAlchemy
from flask import Flask, redirect
from flask_swagger_ui import get_swaggerui_blueprint
from safrs import SAFRSBase, jsonapi_rpc, SAFRSJSONEncoder, Api
from safrs import search, startswith
app = Flask("SAFRS Demo App")
app.config.update(SQLALCHEMY_DATABASE_URI="mysql+pymysql://root:password@localhost/employees", DEBUG=True)
app.url_map.strict_slashes = False
SAFRSBase.db_commit = False
builtins.db = SQLAlchemy(app) # set db as a global variable to be used in employees.py
import employees
def start_api(HOST="0.0.0.0", PORT=80):
with app.app_context():
api = Api(app, api_spec_url="/api/swagger", host="{}:{}".format(HOST, PORT), schemes=["http"], description="")
# Get the SAFRSBase models from employees
for name, model in inspect.getmembers(employees):
bases = getattr(model, "__bases__", [])
if SAFRSBase in bases:
# Create an API endpoint
api.expose_object(model)
department = relationship("Department")
employee = relationship("Employee")
class Salary(SAFRSBase, Base):
__tablename__ = "salaries"
emp_no = Column(ForeignKey("employees.emp_no", ondelete="CASCADE"), primary_key=True, nullable=False)
salary = Column(INTEGER(11), nullable=False)
from_date = Column(Date, primary_key=True, nullable=False)
to_date = Column(Date, nullable=False)
employee = relationship("Employee")
class Title(SAFRSBase, Base):
__tablename__ = "titles"
emp_no = Column(ForeignKey("employees.emp_no", ondelete="CASCADE"), primary_key=True, nullable=False)
title = Column(String(50), primary_key=True, nullable=False)
from_date = Column(Date, primary_key=True, nullable=False)
to_date = Column(Date)
employee = relationship("Employee")
__tablename__ = "payment"
payment_id = Column(SMALLINT(5), primary_key=True)
customer_id = Column(ForeignKey("customer.customer_id", onupdate="CASCADE"), nullable=False, index=True)
staff_id = Column(ForeignKey("staff.staff_id", onupdate="CASCADE"), nullable=False, index=True)
rental_id = Column(ForeignKey("rental.rental_id", ondelete="SET NULL", onupdate="CASCADE"), index=True)
amount = Column(DECIMAL(5, 2), nullable=False)
payment_date = Column(DateTime, nullable=False)
last_update = Column(TIMESTAMP, nullable=False, server_default=text("CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"))
customer = relationship("Customer")
rental = relationship("Rental")
staff = relationship("Staff")
class Rental(SAFRSBase, Base):
__tablename__ = "rental"
__table_args__ = (Index("rental_date", "rental_date", "inventory_id", "customer_id", unique=True),)
rental_id = Column(INTEGER(11), primary_key=True)
rental_date = Column(DateTime, nullable=False)
inventory_id = Column(ForeignKey("inventory.inventory_id", onupdate="CASCADE"), nullable=False, index=True)
customer_id = Column(ForeignKey("customer.customer_id", onupdate="CASCADE"), nullable=False, index=True)
return_date = Column(DateTime)
staff_id = Column(ForeignKey("staff.staff_id", onupdate="CASCADE"), nullable=False, index=True)
last_update = Column(TIMESTAMP, nullable=False, server_default=text("CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"))
customer = relationship("Customer")
inventory = relationship("Inventory")
staff = relationship("Staff")
# Following method is exposed through the REST API
# This means it can be invoked with a HTTP POST
@classmethod
@jsonapi_rpc(http_methods=["POST"])
def send_mail(self, **args):
"""
description : Send an email
args:
email:
type : string
example : test email
"""
return {"result": args}
class Book(SAFRSBase, db.Model):
"""
description: Book description
"""
__tablename__ = "Books"
id = db.Column(db.String, primary_key=True)
name = db.Column(db.String, default="")
user_id = db.Column(db.String, db.ForeignKey("Users.id"))
user = db.relationship("User", back_populates="books")
if __name__ == "__main__":
HOST = sys.argv[1] if len(sys.argv) > 1 else "0.0.0.0"
PORT = 5000
app = Flask("SAFRS Demo Application")
app.config.update(SQLALCHEMY_DATABASE_URI="sqlite://", DEBUG=True)