<jupyter_start><jupyter_text>Python ORMThis notebook will show you how the sqlalchemy library can be used to provide an object relational mapper for python classes and sqlite sql database files.<jupyter_code>pip install --upgrade Perdy
#!/usr/bin/env python3

import os, re, sys, argparse, json, datetime, logging

from io import StringIO

from Perdy.pretty import prettyPrintLn<jupyter_output><empty_output><jupyter_text>sqlalchemyThe most excellent ORM available for python is the sqlalchemy library<jupyter_code>pip install --upgrade sqlalchemy
import sqlalchemy
import sqlalchemy.orm
from sqlalchemy import \
    Column, \
    Integer, \
    String, \
    ForeignKey
from sqlalchemy.engine import \
    reflection
from sqlalchemy.orm import \
    relationship, \
    backref
from sqlalchemy.orm.collections import \
    InstrumentedList, \
    InstrumentedDict, \
    InstrumentedSet, \
    MappedCollection
from sqlalchemy.ext.declarative import \
    declarative_base<jupyter_output><empty_output><jupyter_text>jsonwebto serialise python classes we are using jsonweb decorator, which allows you to easily render python classes in json<jupyter_code>pip install --upgrade jsonweb
from jsonweb.encode import \
    to_object, \
    dumper
from jsonweb.decode import \
    from_object, \
    loader
# the following shippet shows visible items in the dir() of the module
print(json.dumps(sorted(list(filter(lambda x: not x.startswith('_'), dir()))), indent=4))<jupyter_output>[
    "Column",
    "ForeignKey",
    "In",
    "InstrumentedDict",
    "InstrumentedList",
    "InstrumentedSet",
    "Integer",
    "MappedCollection",
    "Out",
    "String",
    "StringIO",
    "argparse",
    "backref",
    "datetime",
    "declarative_base",
    "dumper",
    "exit",
    "from_object",
    "get_ipython",
    "json",
    "loader",
    "logging",
    "os",
    "prettyPrintLn",
    "quit",
    "re",
    "reflection",
    "relationship",
    "sqlalchemy",
    "sys",
    "to_object"
]<jupyter_text>quietlyto hide some of the logging noise lets set the level to warning for those noisy critters<jupyter_code>def quietly():
    for logger in [
        "sqlalchemy.orm.relationships.RelationshipProperty",
        "sqlalchemy.orm.strategies.LazyLoader", "sqlalchemy.orm.path_registry",
        "sqlalchemy.orm.mapper.Mapper", "sqlalchemy.engine.base.Engine",
        "sqlalchemy.pool.NullPool", "sqlalchemy.pool.impl.NullPool",
        "sqlalchemy.pool.SingletonThreadPool",
        "requests.packages.urllib3.connectionpool"
    ]:
        print(logger)
        logging.getLogger(logger).setLevel(logging.WARNING)
        
quietly()<jupyter_output>sqlalchemy.orm.relationships.RelationshipProperty
sqlalchemy.orm.strategies.LazyLoader
sqlalchemy.orm.path_registry
sqlalchemy.orm.mapper.Mapper
sqlalchemy.engine.base.Engine
sqlalchemy.pool.NullPool
sqlalchemy.pool.impl.NullPool
sqlalchemy.pool.SingletonThreadPool
requests.packages.urllib3.connectionpool<jupyter_text>Basesqlalchemy uses a Base class to allow objects to be marshalled and unmarshalled from the database MyBaseto showcase how we can implement inheritance in an SQL ORM, I have chosen to create a MyBase class that stores an ID for the classes, this saves having to do it each time for each sub class. It also shows how we use an "inherited" attribute to define the "type" of the subclass<jupyter_code>Base = declarative_base()

class MyBase(Base):

    __tablename__ = 'base'
    id            = Column(Integer, primary_key=True)
    inherited     = Column(String(256))

    __mapper_args__ = {
        'polymorphic_identity' : 'base',
        'polymorphic_on' : inherited
    }

    def __init__(self,
        id=None,
        inherited=None
    ):
        self.id = id
        self.inherited=inherited
        return

    def __dir__(self):
        return [
            'id'
        ]
    
print(json.dumps(sorted(list(filter(lambda x: not x.startswith('_'), dir(MyBase)))), indent=4))<jupyter_output>[
    "id",
    "inherited",
    "metadata"
]<jupyter_text>Addresslets define a class called Address that extends our polymorphic base class base.idbecause our *Address* class extgends the *MyBase* class, we have to use a *ForeignKey* to link the parent to the child ForeignKeywe are setting up an sql foreign key to be used to link between the Address class and the forthcomming User class. superplease note that we are ensuring to call the super __init__ method of the base class<jupyter_code>@from_object()
@to_object()
class Address(MyBase):

    __tablename__ = 'address'
    id            = Column(Integer,    ForeignKey('base.id'), primary_key=True)
    email         = Column(String(50), nullable=False)
    user_id       = Column(Integer,    ForeignKey('user.id'))

    __mapper_args__ = {
        'polymorphic_identity':'address'
    }

    def __init__(self,
        id=None,
        inherited='address',
        email=None,
        user_id=None
    ):
        super(Address,self).__init__(
            id=id,
            inherited=inherited
        )
        self.email=email
        self.user_id=user_id
        return

    def __dir__(self):
        return [
            'id',
            'email'
        ]
    
print(json.dumps(sorted(list(filter(lambda x: not x.startswith('_'), dir(Address)))), indent=4))<jupyter_output>[
    "email",
    "id",
    "inherited",
    "metadata",
    "user_id"
]<jupyter_text>UserTo show a one to many relationship, we create a User class type that has a *relationship* with the *Address* class.<jupyter_code>@from_object()
@to_object()
class User(MyBase):

    __tablename__ = 'user'
    id            = Column(Integer,         ForeignKey('base.id'), primary_key=True)
    name          = Column(String(50))
    fullname      = Column(String(25))
    password      = Column(String(25))
    addresses     = relationship("Address", uselist=True, foreign_keys=[Address.user_id])

    __mapper_args__ = {
        'polymorphic_identity':'user'
    }

    def __init__(self,
        id=None,
        inherited='user',
        name=None,
        fullname=None,
        password=<PASSWORD>,
        addresses=[]
    ):
        super(User,self).__init__(
            id=id,
            inherited=inherited
        )
        self.name=name
        self.fullname=fullname
        self.password=password
        self.addresses=addresses
        return

    def __dir__(self):
        return [
            'id',
            'name',
            'fullname',
            'password',
            'addresses'
        ]
    
print(json.dumps(sorted(list(filter(lambda x: not x.startswith('_'), dir(User)))), indent=4))<jupyter_output>[
    "addresses",
    "fullname",
    "id",
    "inherited",
    "metadata",
    "name",
    "password"
]<jupyter_text>deployThe *engine* is used to setup an ORM relationship with the classes.To create a database from scratch, we use the *Base.metadata.create_all* method to build a new sqlite database fromt he ORM model.Note we can use any number of database technologies, please see the [sqlalchemy](https://sqlalchemy.org) site for more detail.<jupyter_code>def deploy(url, database, make=False, verbose=False):
    if make:
        if os.path.isfile(database):
            os.unlink(database)

    engine = sqlalchemy.create_engine('%s/%s'%(url,database), echo=verbose)

    if make:
        Base.metadata.create_all(engine)

    return engine<jupyter_output><empty_output><jupyter_text>reflectWe have created a help class to instanciate a Session object that is used to transact with the database, as part of this we have thrown in some reflection allowing you to intraspect the ORM model programatically to see what is already in an existing database.<jupyter_code>def reflect(engine, verbose=False):
    inspector = reflection.Inspector.from_engine(engine)
    if verbose:    
        print(inspector)
        for table in inspector.get_table_names():
            print(table)
            for column in inspector.get_columns(table):
                print('\t%s : %s'%(column['name'], column['type']))

    Session = sqlalchemy.orm.sessionmaker()
    Session.configure(bind=engine)
    return Session<jupyter_output><empty_output><jupyter_text>insertTo insert data we make use of a *session* which we will show later to create a python class object, fill it with values, then add it to the sqlalchemy session and commit the changes.<jupyter_code>def insert(session, verbose=False):
    user = User(
        name='dedson',
        fullname='<NAME>',
        password='<PASSWORD>'
    )
    user.addresses = [
        Address(
            email='<EMAIL>'
        ),
        Address(
            email='<EMAIL>'
        )
    ]
    session.add(user)
    session.commit()
    if verbose:
        print('saved user.id=%s'%user.id)
    return user.id<jupyter_output><empty_output><jupyter_text>queryTo show one of they many ways we can retrieve or query data from the ORM, we make use of the *session* to query based on a *User* class type, and then filter based an attribute name=value pattern.<jupyter_code>def query(session, verbose=False):
    users = session.query(User).filter_by(name='dedson').all()
    js = json.loads(dumper(users))
    if verbose:
        print(json.dumps(js, indent=4))
    return js<jupyter_output><empty_output><jupyter_text>setupNow that we have all the definitions and scaffoldoign in place, lets actually connect to the datavbase and create a Session to play with.<jupyter_code>url = 'sqlite://'
database = 'mySqlAlchemy.db'
print(database)

engine = deploy(url, database, make=True)
Session = reflect(engine, verbose=False)<jupyter_output>mySqlAlchemy.db<jupyter_text>Session()Instantiate a Session object and use it to insert a record.<jupyter_code>session = Session()
insert(session, verbose=True)
session.close()<jupyter_output>saved user.id=1<jupyter_text>queryCreate a new session object and use it to query the database.<jupyter_code>session = Session()
js = query(session, verbose=False)
session.close()

prettyPrintLn(js)<jupyter_output>[36m[[0m
  [35m{[0m
    "[32maddresses[0m" : [36m[[0m
      [35m{[0m
        "[32memail[0m" : "[31me<EMAIL>[0m",
        "[32mid[0m" : [31m2[0m,
        "[32m__type__[0m" : "[31mAddress[0m"
      [35m}[0m,
      [35m{[0m
        "[32memail[0m" : "[31<EMAIL>[0m",
        "[32mid[0m" : [31m3[0m,
        "[32m__type__[0m" : "[31mAddress[0m"
      [35m}[0m
    [36m][0m,
    "[32mfullname[0m" : "[31m<NAME>[0m",
    "[32mid[0m" : [31m1[0m,
    "[32mname[0m" : "[31mdedson[0m",
    "[32mpassword[0m" : "[31m<PASSWORD>[0m",
    "[32m__type__[0m" : "[31mUser[0m"
  [35m}[0m
[36m][0m
{'fraction_non_alphanumeric': 0.11505255255255255, 'fraction_numerical': 0.01144894894894895, 'mean_word_length': 3.307599029911075, 'pattern_counts': {'":': 0, '<': 58, '<?xml version=': 0, '>': 58, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '14', 'dir': 'jupyter-structured-clean-dedup', 'id': '616265', 'license': 'MIT', 'n_tokens_mistral': 3603, 'n_tokens_neox': 3396, 'n_words': 839, 'path': 'mySqlAlchemy.ipynb', 'repo_name': 'eddo888/Carnets', 'size': '10702'}