ORM full name ObjectRelationalMapping: ObjectRelationalMapping. Simply put, ORM creates a correspondence between tables in a database and classes in an object-oriented language expression. That way, if we want to manipulate a database, a table in the database or a record in the table can be done directly from the action class or class case.

SQLAlchemy is one of the best known ORM tools in the Python community, completing a complete enterprise-level persistence model for designing efficient and high-performance database access.

SQLAlchemy advantages:

  1. Concise and easy to read: Abstracts data tables into objects (data models) that are easier to visualize and read.
  2. Portable: encapsulates 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 avoiding SQL injection.

This article introduces the use of SQLAlchemy by introducing common Sqlite database practices. SQLAlchemy is created by transforming database tables into Python classes with data columns as properties and database operations as methods.

Learning materials are free to receive 60+ famous enterprises to share internal resources

SQLAlchem installation

Sqlite3 is the Python3 standard library and does not require an additional installation, just SQLAlchemy.

pip install sqlalchemy
Copy the code

ORM creates database connections

MySQL > create a connection to a database; MySQL > create a connection to a database;

SQLite

Create a database in the current directory in relative path format as follows:

from sqlalchemy import create_engine

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

Create a database as an absolute path in the following format:

from sqlalchemy import create_engine

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

Create database connections for other common databases

SQLAlchemy represents connection information as a string:

'Database type + database driver name :// Username: password @IP address: port number/database name'Copy the code

PostgreSQL database

from sqlalchemy import create_engine # default, The connection string format is "Database type + database driver :// Database username: 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 the code

The MySQL database

From sqlalchemy import create_engine # default, connect string format is "database type + database driver :// database username: 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 the code

The Oracle database

From sqlalchemy import create_engine # default, connect string format is "database type + database driver :// database username: 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 the code

Let’s take creating an SQLite database in the current directory as an example and use this database in the next steps. We added two parameters to the create_engine method. As follows:

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

  • Echo: The default value of echo is False, indicating that detailed execution information such as SQL statements will not be printed. Ture indicates that detailed execution information will be printed.
  • Check_same_thread: Sqlalchemy is multithreaded, so we need to specify check_same_thread=False to make the created object available to any thread.

Define mappings (class to table mappings)

First, we create the base mapping class, from which the concrete mapping class (table) needs to inherit.

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

Next, we create a concrete mapping class. In our case, the Person mapping class is mapped to the Person table.

from sqlalchemy import Column, Integer, String from sqlalchemy. Ext. Declarative import declarative_base Base = declarative_base () # define the mapping class Person, And inherit Base class Person(Base): __tablename__ = 'Person' # If more than one class points to the same table, then the following class needs to set extend_existing to True, which means to extend the existing column # or in other words, Sqlalchemy allows classes to be table word sets like this: # __table_args__ = {'extend_existing': If the table is in the same database service (datebase), you can use the schema parameter to further specify the database # __table_args__ = {'schema': 'AiTestOps_database'} # sqlalchemy will select a primary key (s) from the database where the query is performed. Set the variable names to be the same as the table fields, as the same name is the only association between them. Specify person_id to map to person_id; The person_id field is an integer, the primary key, Person_id = Column(Integer, primary_key=True, autoincrement=True) # map username to username; The username field is a string.  username = Column(String(20), nullable=False, Index =True) password = Column(String(32)) desc = Column(String(32)) # __repr__ __repr__(self): return "<User(username='%s', password='%s', desc='%s')>" % ( self.username, self.password, self.desc)Copy the code

First, to be clear, ORM tables do not normally need to exist first. As we saw in the Person class, tablename is used to specify the name of the table in SQLite.

We create three fields in Person, each Column in the class represents a Column (field) in the database, and in Colunm, specify some attributes for that Column. The first field represents the data type. Above we use String and Integer as the two most common types. Other common types include Text, Boolean, SmallInteger, and DateTime.

Nullable =False indicates that the column cannot be null, and index=True indicates that the column is indexed. In addition, the REPR is defined for debugging purposes.

In the Person class mapping definition above, the __tablename__ attribute is static, but sometimes we might want to pass the tablename to the class dynamically from the outside. This can be done by defining the 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, Class User_Model(Base) =declarative_base()) __tablename__ = table_name __table_args__ = {'extend_existing': True} person_id = Column(Integer, primary_key=True, autoincrement=True) The username field is a string.  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 User_Model if __name__ == '__main__': TestModel = table_name_model_class("Person_Info") print(TestModel.__table__)Copy the code

Create table

View the table corresponding to the mapping

Person.__table__
Copy the code

Create tables for all classes that inherit from Base

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

Checkfirst The default value is True, which indicates that the table is checked before being created. If the table with the same name already exists, the table will not be created.

Create a specified table

Base.metadata.create_all(engine, tables=[Base.metadata.tables['Person']], Person.__table__. Create (engine, checkFirst =True)Copy the code

Establish a session

Orm import sessionmaker from sqlalchemy import create_engine # create link engine = create_engine(r'sqlite:///AiTestOps.db? check_same_thread=False', Session = sessionmaker(bind=engine)Copy the 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()): Class User_Model(Base): # Assign tablename __tablename__ = table_name __table_args__ = {'extend_existing': True} person_id = Column(Integer, primary_key=True, autoincrement=True) The username field is a string.  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 User_Model if __name__ == '__main__': Person = table_name_model_class("Person") # create link engine = create_engine(r'sqlite:///AiTestOps.db?check_same_thread=False', Echo =True) # create table person.__table__. Create (engine, Session = sessionmaker(bind=engine) Session = sessionmaker(bind=engine jon_info = Person(username='Jon', password='123456', Add_all ([Person(username='Mark', name='Mark', name='Mark', name='Mark', name='Mark', name='Mark', name='Mark', name='Mark')) Password ='123456', desc=' lively '), Person(username='Tony', password='123456', desc=' lively ')]) Session.mit () requires a commit to confirm that the changes are written to the database.Copy the 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()): Class User_Model(Base): # Assign tablename __tablename__ = table_name __table_args__ = {'extend_existing': True} person_id = Column(Integer, primary_key=True, autoincrement=True) The username field is a string.  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 User_Model if __name__ == '__main__': Person = table_name_model_class("Person") # create link engine = create_engine(r'sqlite:///AiTestOps.db?check_same_thread=False', Echo =True) # create table person.__table__. Create (engine, Session = sessionmaker(bind=engine) Session = sessionMaker (bind=engine) Session = sessionMaker ( Session. add_all([Person(username='Mark', password='123456', desc=' liveliness '), Person(username='Tony', password='123456', Select * from session.mit (); select * from session.mit () where username='Mark'; Mark = session.query(Person).filter_by(username=' mark ').all() print(mark) # Mark_desc = session.query(person.desc).filter_by(username='Mark').all() print(mark_desc)Copy the code

To better understand the difference between SQL and SQLalchemy, see the following:

  • Query: corresponds to SELECT XXX FROM XXX
  • Filter /filter_by: indicates WHERE fillter can perform comparison operations (==, >, <…). Different conditions are separated by commas. Fillter_by can only specify parameters to pass to obtain the query result.
  • Limit: corresponds to limit()
  • Order by: corresponds to order_by()
  • Group by: corresponding to group_by()

Like the query

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

Is the query

# is_ == result = session.query(Person).filter(person.username. Is_ (None)).all() result = Session.query (Person).filter(person.username == None).all() # isnot =! = result = session.query(Person).filter(Person.username.isnot(None)).all() result = session.query(Person).filter(Person.username ! = None).all()Copy the code

Regular query

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

Statistics of the number

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

IN the query

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

NOT IN the query

Not more_person = session.query(Person).filter(~ person.username. In_ (['Mark', 'Tony']).all() # notin_ more_person = session.query(Person).filter(~ person.username. Notin_ (['Mark', 'Tony'])).all() # notin_ more_person = session.query(Person).filter(~ person.username.Copy the code

AND the query

from sqlalchemy import and_ more_person = Session. The query (Person). The filter (and_ (Person. Password = = '123456', the Person the desc = = "cute")) all ()Copy the code

The OR query

from sqlalchemy import or_ more_person = Session. The query (Person). The filter (or_ (Person. Password = = '123456', the Person the desc = = "lively")) all ()Copy the code

Grouping query

Std_group_by = session.query(Person).group_by(person.desc).all() # from sqlalchemy. SQL import func res = Session.query (person.desc, func.count(person.desc),).group_by(person.desc).all() # print(person)Copy the code

Sorting query

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

Limit the query

If the number of arguments in the limit table is greater than the number of arguments in the instance table, the number of arguments in the limit table is greater than the number of arguments in the instance table. Data_limit = session.query(Person).filter(person.desc.notLike (" live %")).limit(1).all() data_limit = session.query(Person).filter(person.desc.notLike (" live %")).limit(1).all()Copy the code

Offset query

Select offset from table where offset is an integer. Offset (1).all() result = data_like = session.query(Person).filter(person.desc.like (" live %")).offset(1).all() result = session.query(Person).offset(1).limit(6).all()Copy the code

Aggregation function

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 the code

About the number of results returned

All () - Query all - returns a list object first() - Queries the first eligible object - returns an objectCopy the code

About the reference

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

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

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()): Class User_Model(Base): # Assign tablename __tablename__ = table_name __table_args__ = {'extend_existing': True} person_id = Column(Integer, primary_key=True, autoincrement=True) The username field is a string.  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 User_Model if __name__ == '__main__': Person = table_name_model_class("Person") # create link engine = create_engine(r'sqlite:///AiTestOps.db?check_same_thread=False', Echo =True) # create table person.__table__. Create (engine, Session = sessionmaker(bind=engine) Session = sessionMaker (bind=engine) Session = sessionMaker ( Session. add_all([Person(username='Mark', password='123456', desc=' liveliness '), Person(username='Tony', password='123456', Desc =' active ')]) # desc=' active ') Session.mit () person = session.query(person).filter_by(username='Mark').first() Session.mit () our_user = '654321 session.query(Person.password).filter_by(username='Mark').all() print(our_user)Copy the code

The above operation, first query and then modify, equivalent to executing two statements, we can directly use the following method

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

Update table synchronize_session=False; update/delete table synchronize_session=False;

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

Update a table from one schema to another. This is the same as updating a table from the same schema, except that when defining the model, you need to specify the corresponding schema in table_args = {‘schema’: ‘test_Person’}.

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()): Class User_Model(Base): # Assign tablename __tablename__ = table_name __table_args__ = {'extend_existing': True} person_id = Column(Integer, primary_key=True, autoincrement=True) The username field is a string.  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 User_Model if __name__ == '__main__': Person = table_name_model_class("Person_Info") # create link engine = create_engine(r'sqlite:///AiTestOps.db?check_same_thread=False', Echo =True) # create table person.__table__. Create (engine, Session = sessionmaker(bind=engine) Session = sessionMaker (bind=engine) Session = sessionMaker ( Session. add_all([Person(username='Mark', password='123456', desc=' liveliness '), Person(username='Tony', password='123456', Desc =' active ')]) # desc=' active ') Session.mit () mark = session.query(Person).filter_by(username=' mark ').first() # Session.delete (mark) # delete session.mit () # delete session.mit () # delete session.mit () # delete session.mit () # delete session.mit () print(person.username)Copy the code

Or, directly in one step, do not need to query out, and then delete the operation as above.

Software test materials are here!! A treasure trove of test learning.

Session.query (Person).filter(person.username == "Mark").delete() session.com MIT () # delete in Synchronize_session =False; Otherwise would be thrown qlalchemy. Exc. InvalidRequestError session. The query (Person). The filter (Person) desc) in_ ([' cute ' 'lively'])). The delete (synchronize_session = False)Copy the code

​