ORM Full name ObjectRelationalMapping, Object relation mapping . In short ,ORM This paper creates a class correspondence between the tables in the database and the classes expressed in object-oriented language . like that , We need to operate the database , A table in the database or a record in the table can be completed directly according to the operation class or class case .

SQLAlchemy yes Python The most famous ORM One of the tools , Design solutions for efficient and performance database access , Completed a complete enterprise persistence model .

SQLAlchemy advantage ：

1. Simple and easy to read ： Abstract a data table as an object （ Data model ）, More visual and easy to read .
2. portable ： Encapsulate a variety of database engines , Dealing with multiple databases , The actual operation is basically the same , The code is easy to maintain .
3. More secure ： Effectively avoid SQL Inject .

Through the introduction of this article Sqlite Common practical operations of database , Let's introduce SQLAlchemy How to use .SQLAlchemy The specific establishment method is to transform the database table into Python class , Where data columns are properties , Database operations as a method .

Learning materials are free 60+ Famous enterprises promote resource sharing

SQLAlchem install

Sqlite3 yes Python3 The standard library does not need to be installed separately , Just install SQLAlchemy that will do .

pip install sqlalchemy
 Copy code 

---

ORM Create database connection

Sqlite3 To create a database connection is to create a database , Others MySQL Such as the database , The required database already exists , To create a database connection .

SQLite

In the form of a relative path , Create a database in the current directory in the following format ：

from sqlalchemy import create_engine

engine = create_engine('sqlite:///AiTestOps.db')
 Copy code 

Create the database as an absolute path , The format is as follows ：

from sqlalchemy import create_engine

engine = create_engine('sqlite:///G:\python_sql\AiTestOps.db')
 Copy code 

Other common database creation and database connection methods

SQLAlchemy Use a string to represent connection information ：

' Database type + Database driver name :// user name : password @IP Address : Port number / Database name '
 Copy code 

PostgreSQL database

from sqlalchemy import create_engine

# default,  The connection string format is  " Database type + Database driven :// Database user name : Database password @IP Address : port / database "
engine = create_engine('postgresql://username:[email protected]:9527/AiTestOps')
# psycopg2
engine = create_engine('postgresql+psycopg2://username:[email protected]:9527/AiTestOps')
# pg8000
engine = create_engine('postgresql+pg8000://username:[email protected]:9527/AiTestOps')
 Copy code 

MySQL database

from sqlalchemy import create_engine

# default, The connection string format is  " Database type + Database driven :// Database user name : Database password @IP Address : port / database "
engine = create_engine('mysql://username:[email protected]:9527/AiTestOps')
# mysql-python
engine = create_engine('mysql+mysqldb://username:[email protected]:9527/AiTestOps')
# MySQL-connector-python
engine = create_engine('mysql+mysqlconnector://username:[email protected]:9527/AiTestOps')
 Copy code 

Oracle database

from sqlalchemy import create_engine

# default, The connection string format is  " Database type + Database driven :// Database user name : Database password @IP Address : port / database "
engine = create_engine('oracle://username:[email protected]:9527/AiTestOps')
# cx_oracle
engine = create_engine('oracle+cx_oracle://username:[email protected]:9527/AiTestOps')
 Copy code 

Let's create... In the current directory SQLite Database, for example , The following steps are the same as using this database . We are create_engine Method is supplemented with two parameters . as follows ：

from sqlalchemy import create_engine

engine = create_engine('sqlite:///AiTestOps.db?check_same_thread=False', echo=True)
 Copy code 

• echo：echo The default is False, Indicates that... Is not printed SQL Statement and other detailed execution information , Change it to Ture Means to let it print .
• check_same_thread：check_same_thread The default is False,sqlite The default created object can only be used by the thread that created the object , and sqlalchemy It's multi-threaded , So we need to specify check_same_thread=False To make the established object available to any thread .

---

Define mapping （ Mapping between class and table ）

First , We build the basic mapping class , The following specific mapping class （ surface ） Need to inherit it .

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
 Copy code 

then , Create a concrete mapping class , We are here to Person Take mapping class as an example , We put Person Class to Person surface .

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

#  Define mapping classes Person, And inheritance  Base
class Person(Base):
    #  Specifies that this class is mapped to  Person  surface 
    __tablename__ = 'Person'
    #  If there are multiple classes pointing to the same table , Then the class in the back needs to put  extend_existing Set to True, It means to extend the existing columns 
    #  Or to put it another way ,sqlalchemy  The allowed class is the word set of the table , as follows ：
    # __table_args__ = {'extend_existing': True}
    #  If tables are in same database service （datebase） In different databases （schema）, You can use schema Parameter further specifies the database 
    # __table_args__ = {'schema': 'AiTestOps_database'}

    # sqlalchemy  It is mandatory to have a primary key field, otherwise an error will be reported ,sqlalchemy After receiving the query results, you will perform a de duplication according to the primary key , Therefore, do not set the non primary key field to primary_key
    #  Each variable name must be the same as each field name of the table , Because the same name is the only relationship between them , Appoint  person_id  Mapping to  person_id  Field ; person_id  The field is an integer , Primary key , Automatic growth （ In fact, the integer primary key grows automatically by default ）
    person_id = Column(Integer, primary_key=True, autoincrement=True)
    #  Appoint  username  Mapping to  username  Field ; username  The field is a string type ,
    #  Appoint  username  Mapping to  username  Field ; username  The field is a string type ,
    username = Column(String(20), nullable=False, index=True)
    password = Column(String(32))
    desc = Column(String(32))

    # __repr__ Method is used to output the object of this class print() String output when 
    def __repr__(self):
        return "<User(username='%s', password='%s', desc='%s')>" % (
            self.username, self.password, self.desc)
 Copy code 

First of all, make it clear ,ORM In general, tables do not need to exist first , We see , stay Person Class , use tablename Specified in the SQLite The name of the middle watch .

We are Person Three fields are created in , Each of the classes Column Represents a column in a database （ Field ）, stay Colunm in , Specify some properties of the column . The first field represents the data type , We use String, Integer The two most common types , Other common ones include ：Text、Boolean、SmallInteger、DateTime.

nullable=False Represents that this column cannot be empty ,index=True Means to create an index in this column . in addition , Definition repr For the convenience of debugging .

Above Person Class mapping definition ,__tablename__ Property is static , But sometimes we may want to dynamically pass the table name to the class through the outside , At this point, you can define an internal class to pass parameters , as follows ：

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

def table_name_model_class(table_name, Base=declarative_base()):
    #  Define an inner class 
    class User_Model(Base):
        #  Assign a value to the table name 
        __tablename__ = table_name
        __table_args__ = {'extend_existing': True}

        person_id = Column(Integer, primary_key=True, autoincrement=True)
        #  Appoint  username  Mapping to  username  Field ; username  The field is a string type ,
        username = Column(String(20), nullable=False, index=True)
        password = Column(String(32))
        desc = Column(String(32))
        
        def __repr__(self):
            return "<User(username='%s', password='%s', desc='%s')>" % (
                self.username, self.password, self.desc)
    #  Return the class that dynamically sets the table name to 
    return User_Model

if __name__ == '__main__':
    TestModel = table_name_model_class("Person_Info")
    print(TestModel.__table__)
 Copy code 

---

Create data table

View the table corresponding to the mapping

Person.__table__
 Copy code 

Create all inherited from Base The table corresponding to the class of

Base.metadata.create_all(engine, checkfirst=True)
 Copy code 

checkfirst The default is True, Indicates to check whether the table exists before creating it , If a table with the same name already exists , No more creation of .

Create the specified table

Base.metadata.create_all(engine, tables=[Base.metadata.tables['Person']], checkfirst=True)
#  Or is it 
Person.__table__.create(engine, checkfirst=True)
 Copy code 

---

Establish a conversation

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
#  Create links  
engine = create_engine(r'sqlite:///AiTestOps.db?check_same_thread=False', echo=True)
#  establish Session Class object 
Session = sessionmaker(bind=engine)
#  establish Session Class instance 
session = Session()
 Copy code 

---

insert data

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine

def table_name_model_class(table_name, Base = declarative_base()):
    #  Define an inner class 
    class User_Model(Base):
        #  Assign a value to the table name 
        __tablename__ = table_name
        __table_args__ = {'extend_existing': True}

        person_id = Column(Integer, primary_key=True, autoincrement=True)
        #  Appoint  username  Mapping to  username  Field ; username  The field is a string type ,
        username = Column(String(20))
        password = Column(String(32))
        desc = Column(String(32))
        
        def __repr__(self):
            return "<User(username='%s', password='%s', desc='%s')>" % (
                self.username, self.password, self.desc)
    #  Return the class that dynamically sets the table name to 
    return User_Model

if __name__ == '__main__':

    Person = table_name_model_class("Person")
    #  Create links 
    engine = create_engine(r'sqlite:///AiTestOps.db?check_same_thread=False', echo=True)
    #  establish  Person  surface 
    Person.__table__.create(engine, checkfirst=True)
    #  establish Session Class object 
    Session = sessionmaker(bind=engine)
    #  establish Session Class instance 
    session = Session()
    #  establish User Class instance 
    jon_info = Person(username='Jon', password='123456', desc=' lively ')

    #  Insert the instance into  Person  surface 
    session.add(jon_info)

    #  Insert multiple records at a time 
    session.add_all(
        [
        Person(username='Mark', password='123456', desc=' lively '),
        Person(username='Tony', password='123456', desc=' lively ')
        ]
    )

    #  The current change is only in session in , Need to use commit Confirm the changes before writing to the database 
    session.commit()
 Copy code 

---

Query data

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine

def table_name_model_class(table_name, Base = declarative_base()):
    #  Define an inner class 
    class User_Model(Base):
        #  Assign a value to the table name 
        __tablename__ = table_name
        __table_args__ = {'extend_existing': True}

        person_id = Column(Integer, primary_key=True, autoincrement=True)
        #  Appoint  username  Mapping to  username  Field ; username  The field is a string type ,
        username = Column(String(20))
        password = Column(String(32))
        desc = Column(String(32))
        
        def __repr__(self):
            return "<User(username='%s', password='%s', desc='%s')>" % (
                self.username, self.password, self.desc)
    #  Return the class that dynamically sets the table name to 
    return User_Model

if __name__ == '__main__':

    Person = table_name_model_class("Person")
    #  Create links 
    engine = create_engine(r'sqlite:///AiTestOps.db?check_same_thread=False', echo=True)
    #  establish  Person  surface 
    Person.__table__.create(engine, checkfirst=True)
    #  establish Session Class object 
    Session = sessionmaker(bind=engine)
    #  establish Session Class instance 
    session = Session()
    #  Insert multiple records at a time 
    session.add_all(
        [
        Person(username='Mark', password='123456', desc=' lively '),
        Person(username='Tony', password='123456', desc=' lively ')
        ]
    )
    #  The current change is only in session in , Need to use commit Confirm the changes before writing to the database 
    session.commit()
    
    #  Inquire about  username='Mark'  All the results of , Return result object 
    mark = session.query(Person).filter_by(username='Mark').all()
    print(mark)

    #  If you only get some fields , Then the tuple is returned instead of the object 
    mark_desc = session.query(Person.desc).filter_by(username='Mark').all()
    print(mark_desc)
 Copy code 

For better understanding SQL And SQLalchemy The difference in writing , You can refer to the following ：

• query ： Corresponding SELECT xxx FROM xxx
• filter/filter_by ： Corresponding WHERE ,fillter  You can do a comparison (==, >, < ...) To make flexible use of conditions , The different conditions are separated by commas ,fillter_by  You can only specify parameters and pass parameters to get query results .
• limit ： Corresponding limit()
• order by ： Corresponding order_by()
• group by ： Corresponding group_by()

like Inquire about

# like 
data_like = session.query(Person).filter(Person.desc.like(" live %")).all()
# not like
data_like = session.query(Person).filter(Person.desc.notlike(" live %")).all()
 Copy code 

is Inquire about

# is_  amount to  ==
result = session.query(Person).filter(Person.username.is_(None)).all()
result = session.query(Person).filter(Person.username == None).all()
# isnot  amount to  !=
result = session.query(Person).filter(Person.username.isnot(None)).all()
result = session.query(Person).filter(Person.username != None).all()
 Copy code 

Regular queries

data_regexp = session.query(Person).filter(Person.password.op("regexp")(r"^[\u4e00-\u9fa5]+")).all()
 Copy code 

Statistical quantity

data_like_count = session.query(Person).filter(Person.desc.like(" live %")).count()
 Copy code 

IN Inquire about

more_person = session.query(Person).filter(Person.username.in_(['Mark', 'Tony'])).all()
 Copy code 

NOT IN Inquire about

# ~ Represents reverse , convert to sql It's keywords not
more_person = session.query(Person).filter(~Person.username.in_(['Mark', 'Tony'])).all()
#  or  notin_
more_person = session.query(Person).filter(~Person.username.notin_(['Mark', 'Tony'])).all()
 Copy code 

AND Inquire about

from sqlalchemy import and_

more_person = session.query(Person).filter(and_(Person.password=='123456',Person.desc==" lovely '")).all()
 Copy code 

OR Inquire about

from sqlalchemy import or_

more_person = session.query(Person).filter(or_(Person.password=='123456',Person.desc==" lively '")).all()
 Copy code 

Group query

std_group_by = session.query(Person).group_by(Person.desc).all()
#  or 
from sqlalchemy.sql import func

res = session.query(Person.desc,
                    func.count(Person.desc),
                   ).group_by(Person.desc).all()

#  Traversal view , No more ed User record 
for person in res:
    print(person)
 Copy code 

Sort query

std_order_by = session.query(Person).order_by(Person.username.desc()).all()
 Copy code 

limit Inquire about

# limit  Limit quantity query , limit An integer is passed in to restrict the number of views ,  When limit When the parameters in the instance table are greater than the number in the instance table , All query results will be returned 
data_limit = session.query(Person).filter(Person.desc.notlike(" live %")).limit(1).all()
 Copy code 

Offset query

# offset  Offset query ,offset Pass in an integer , Start the query from this location in the table ,offset You can talk to limit Mix to limit 
data_like = session.query(Person).filter(Person.desc.like(" live %")).offset(1).all()
result = session.query(Person).offset(1).limit(6).all()
 Copy code 

Aggregate functions

from sqlalchemy import func, extract
# count
result = session.query(Person.password, func.count(Person.id)).group_by(Person.password).all()
# sum
result = session.query(Person.password, func.sum(Person.id)).group_by(Person.password).all()
# max
result = session.query(Person.password, func.max(Person.id)).group_by(Person.password).all()
# min
result = session.query(Person.password, func.min(Person.id)).group_by(Person.password).all()
# having
result = session.query(Person.password, func.count(Person.id)).group_by(Person.password).having(func.count(Person.id) > 1).all()
 Copy code 

About the number of returned results

all()
-  Query all 
-  Returns a list object 

first()
-  Query the first eligible object 
-  Return an object 
 Copy code 

About biography

filter = (Person.username=='Mark')

our_user = session.query(Person).filter(filter).first()
print(our_user)
 Copy code 

---

to update

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine

def table_name_model_class(table_name, Base = declarative_base()):
    #  Define an inner class 
    class User_Model(Base):
        #  Assign a value to the table name 
        __tablename__ = table_name
        __table_args__ = {'extend_existing': True}

        person_id = Column(Integer, primary_key=True, autoincrement=True)
        #  Appoint  username  Mapping to  username  Field ; username  The field is a string type ,
        username = Column(String(20))
        password = Column(String(32))
        desc = Column(String(32))
        
        def __repr__(self):
            return "<User(username='%s', password='%s', desc='%s')>" % (
                self.username, self.password, self.desc)
    #  Return the class that dynamically sets the table name to 
    return User_Model

if __name__ == '__main__':

    Person = table_name_model_class("Person")
    #  Create links 
    engine = create_engine(r'sqlite:///AiTestOps.db?check_same_thread=False', echo=True)
    #  establish  Person  surface 
    Person.__table__.create(engine, checkfirst=True)
    #  establish Session Class object 
    Session = sessionmaker(bind=engine)
    #  establish Session Class instance 
    session = Session()
    #  Insert multiple records at a time 
    session.add_all(
        [
            Person(username='Mark', password='123456', desc=' lively '),
            Person(username='Tony', password='123456', desc=' lively ')
        ]
    )
    #  The current change is only in session in , Need to use commit Confirm the changes before writing to the database 
    session.commit()
    
    #  To modify, you need to find out the record first 
    person = session.query(Person).filter_by(username='Mark').first()
    #  take  Mark  The user's password is changed to  654321
    person.password = '654321'
    #  Confirm the change 
    session.commit()
    
    our_user = session.query(Person.password).filter_by(username='Mark').all()
    print(our_user)
 Copy code 

The operation above , Query first and then modify , It is equivalent to executing two statements , We can directly use the following methods

session.query(Person).filter_by(username='Mark').update({Person.password: '6543210'})
session.commit()
 Copy code 

With the same schema One table of updates the writing method of another table , Across tables update/delete And so on synchronize_session=False, Otherwise, the report will be wrong ：

session.query(Person).filter_by(Person.username=Person1.username).update({Person.password: Person1.password}, synchronize_session=False)
session.commit()
 Copy code 

With a schema Update another table schema How to write your table , Write in the same way as schema The same as , It's just a definition model When needed table_args = {'schema': 'test_Person'} And so on schema.

---

Delete

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine

def table_name_model_class(table_name, Base = declarative_base()):
    #  Define an inner class 
    class User_Model(Base):
        #  Assign a value to the table name 
        __tablename__ = table_name
        __table_args__ = {'extend_existing': True}

        person_id = Column(Integer, primary_key=True, autoincrement=True)
        #  Appoint  username  Mapping to  username  Field ; username  The field is a string type ,
        username = Column(String(20))
        password = Column(String(32))
        desc = Column(String(32))
        
        def __repr__(self):
            return "<User(username='%s', password='%s', desc='%s')>" % (
                self.username, self.password, self.desc)
    #  Return the class that dynamically sets the table name to 
    return User_Model


if __name__ == '__main__':
    Person = table_name_model_class("Person_Info")

    #  Create links 
    engine = create_engine(r'sqlite:///AiTestOps.db?check_same_thread=False', echo=True)
    #  establish  Person  surface 
    Person.__table__.create(engine, checkfirst=True)
    #  establish Session Class object 
    Session = sessionmaker(bind=engine)
    #  establish Session Class instance 
    session = Session()
    #  Insert multiple records at a time 
    session.add_all(
        [
            Person(username='Mark', password='123456', desc=' lively '),
            Person(username='Tony', password='123456', desc=' lively ')
        ]
    )
    #  The current change is only in session in , Need to use commit Confirm the changes before writing to the database 
    session.commit()

    mark = session.query(Person).filter_by(username='Mark').first()

    #  take  mark  User record deletion 
    session.delete(mark)

    #  confirm deletion 
    session.commit()

    #  Traversal view , No more  Mark  data  
    for person in session.query(Person):
        print(person.username)
 Copy code 

perhaps , One step at a time , You don't need to be like that , Find out first , Then delete .

The software test data are here ！！ A treasure trove of test learning .

session.query(Person).filter(Person.username == "Mark").delete()

session.commit()
#  Delete  in  Operate the query records , Need to transmit synchronize_session=False, Otherwise it will throw  qlalchemy.exc.InvalidRequestError
session.query(Person).filter(Person.desc.in_([' lovely ', ' lively '])).delete(synchronize_session=False) 
 Copy code 

​