Examples

Example files are included that provide working examples of descriptions in the User Guide. They are found in the examples directory on Github at https://github.com/sidorof/dbbase.

• Serialization and Relationships

• Serialization and Self-Referential Relationships

• Documentation Dictionaries

Serialization and Relationships

# examples/two_tables.py
"""
This example mirrors the user guide example for serialization with
two tables.
"""
import json
from dbbase import DB

db = DB(config=":memory:")


class User(db.Model):
    __tablename__ = "users"
    id = db.Column(db.Integer, primary_key=True)
    first_name = db.Column(db.String(50), nullable=False)
    last_name = db.Column(db.String(50), nullable=False)
    addresses = db.relationship("Address", backref="user", lazy="immediate")

    def full_name(self):
        return "{first} {last}".format(
            first=self.first_name, last=self.last_name
        )


class Address(db.Model):
    __tablename__ = "addresses"
    id = db.Column(db.Integer, primary_key=True)
    email_address = db.Column(db.String, nullable=False)
    user_id = db.Column(db.Integer, db.ForeignKey("users.id"))


db.create_all()

user = User(id=3, first_name="Bob", last_name="Smith")
user.save()

address1 = Address(email_address="email1@example.com", user_id=user.id)
address2 = Address(email_address="email2@example.com", user_id=user.id)

db.session.add(address1)
db.session.add(address2)
db.session.commit()

# ------------------------------------------------
print("default serialization for user")
print(user.serialize(indent=2))
input("ready")

# ------------------------------------------------
print("modified serialization: stop list for first/last names")
User.SERIAL_STOPLIST = ["first_name", "last_name"]
print(user.serialize(indent=2))
input("ready")

# ------------------------------------------------
print("now only the email_address portion of address shows")
Address.SERIAL_FIELDS = ["email_address"]
print(user.serialize(indent=2))
input("ready")

# ------------------------------------------------

print("Now ad hoc variables")
print(
    "user serial_fields is {}".format(
        "['id', 'first_name', 'last_name', 'addresses']"
    )
)
print(
    "user serial_field_relations is {}".format(
        "{'Address': ['id', 'email_address'] "
    )
)
print(
    user.serialize(
        indent=2,
        sort=True,
        serial_fields=["id", "first_name", "last_name", "addresses"],
        serial_field_relations={"Address": ["id", "email_address"]},
    )
)
input("ready")
print()
print("serialization of just an address")
print(
    address1.serialize(
        indent=2,
        sort=True,
        serial_fields=["id", "email_address"],
        serial_field_relations={"User": ["user_id", "first_name", "last_name"]},
    )
)

input("ready")

# ------------------------------------------------
print('Print table documentation')
print(json.dumps(db.doc_tables(), indent=4))

Serialization and Self-Referential Relationships

# examples/self_referential.py
"""
This example shows an example where serialization works with
self-referential models.
"""
from dbbase import DB

db = DB(config=":memory:")


class Node(db.Model):
    """self-referential table"""

    __tablename__ = "nodes"
    id = db.Column(db.Integer, primary_key=True)
    parent_id = db.Column(db.Integer, db.ForeignKey("nodes.id"))
    data = db.Column(db.String(50))
    children = db.relationship(
        "Node", lazy="joined", order_by="Node.id", join_depth=10
    )

    SERIAL_FIELDS = ["id", "parent_id", "data", "children"]


db.create_all()

node1 = Node(id=1, data="this is node1")
node2 = Node(id=2, data="this is node2")
node3 = Node(id=3, data="this is node3")
node4 = Node(id=4, data="this is node4")
node5 = Node(id=5, data="this is node5")
node6 = Node(id=6, data="this is node6")

db.session.add(node1)
db.session.commit()
node1.children.append(node2)
db.session.commit()
node2.children.append(node3)
db.session.commit()
node2.children.append(node4)
db.session.commit()
node1.children.append(node5)
db.session.commit()
node5.children.append(node6)
db.session.commit()

print("Example of self-referential serialization")
print(node1.serialize(indent=2))
input("ready")

print("Example in compact form")
print(node1.serialize())

Documentation Dictionaries

# examples/table_documentation.py
"""
This example mirrors the user guide example for documentation
dictionaries.

This is simply a table with different kinds of columns and parameter choices.
The StatusCodes class is included as a method for illustrating a selection
of choices that would be converted to a status code. There are several ways
to do this, this is one of several.

BigTable is the class is created. The function `db.doc_table` creates the
output.

For clarity, the output converted to JSON format.

`print(json.dumps(db.doc_table(BigTable), indent=4))`

"""
import json
from datetime import date, datetime
import sqlalchemy.types as types

from dbbase import DB

db = DB(config=":memory:")

status_codes = [
    [0, "New"],
    [1, "Active"],
    [2, "Suspended"],
    [3, "Inactive"],
]


class StatusCodes(types.TypeDecorator):
    """
    Status codes are entered as strings and converted to
    integers when saved to the database.
    """

    impl = types.Integer

    def __init__(self, status_codes, **kw):
        self.choices = dict(status_codes)
        super(StatusCodes, self).__init__(**kw)

    def process_bind_param(self, value, dialect):
        """called when saving to the database"""
        return [k for k, v in self.choices.items() if v == value][0]

    def process_result_value(self, value, dialect):
        """called when pulling from database"""
        return self.choices[value]


class BigTable(db.Model):
    """Test class with a variety of column types"""

    __tablename__ = "big_table"

    id = db.Column(
        db.Integer,
        primary_key=True,
        nullable=True,
        comment="Primary key with a value assigned by the database",
        info={"extra": "info here"},
    )

    status_id = db.Column(
        StatusCodes(status_codes),
        nullable=False,
        comment="Choices from a list. String descriptors "
        "change to integer upon saving. Enums without the headache.",
    )

    @db.orm.validates("status_id")
    def _validate_id(self, key, id):
        """_validate_id

        Args:
            id: (int) : id must be in cls.choices
        """
        if id not in dict(status_codes):
            raise ValueError("{} is not valid".format(id))
        return id

    # nullable / not nullable
    name1 = db.Column(
        db.String(50), nullable=False, comment="This field is required"
    )
    name2 = db.Column(
        db.String(50), nullable=True, comment="This field is not required",
    )

    # text default
    name3 = db.Column(
        db.Text,
        default="test",
        nullable=False,
        comment="This field has a default value",
        index=True,
    )

    item_length = db.Column(
        db.Float, nullable=False, comment="This field is a float value"
    )

    item_amount = db.Column(db.Numeric(17, 6), default=0.0)

    # integer and default
    some_small_int = db.Column(
        db.SmallInteger,
        default=0,
        nullable=False,
        comment="This field is a small integer",
    )
    some_int = db.Column(
        db.Integer,
        default=0,
        nullable=False,
        comment="This field is a 32 bit integer",
    )
    some_big_int = db.Column(
        db.BigInteger,
        default=0,
        nullable=False,
        comment="This field is a big integer",
    )

    fk_id = db.Column(
        db.Integer,
        db.ForeignKey("other_table.id"),
        nullable=False,
        comment="This field is constrained by a foreign key on"
        "another table",
    )

    today = db.Column(
        db.Date,
        doc="this is a test",
        info={"test": "this is"},
        comment="This field defaults to today, created at model level",
        default=date.today,
    )

    created_at1 = db.Column(
        db.DateTime,
        default=datetime.now,
        comment="This field defaults to now, created at model level",
    )
    created_at2 = db.Column(
        db.DateTime,
        server_default=db.func.now(),
        comment="This field defaults to now, created at the server" "level",
    )
    update_time1 = db.Column(
        db.DateTime,
        onupdate=datetime.now,
        comment="This field defaults only on updates",
    )
    update_time2 = db.Column(
        db.DateTime,
        server_onupdate=db.func.now(),
        comment="This field defaults only on updates, but on the" "server",
    )

    unique_col = db.Column(
        db.String(20),
        unique=True,
        comment="This must be a unique value in the database.",
    )

    # adapted from sqlalchemy docs
    abc = db.Column(
        db.String(20),
        server_default="abc",
        comment="This field defaults to text but on the server.",
    )
    index_value = db.Column(
        db.Integer,
        server_default=db.text("0"),
        comment="This field defaults to an integer on the server.",
    )

    __table_args = (db.Index("ix_name1_name2", "name1", "name2", unique=True),)


class OtherTable(db.Model):
    """
    This table is used solely to enable an option for a foreign key.
    """

    __tablename__ = "other_table"
    id = db.Column(db.Integer, primary_key=True, nullable=True)


# using JSON for a better output

print(json.dumps(db.doc_table(BigTable), indent=4))