current position:Home>Database addition, deletion, modification and query of Python Sqlalchemy basic operation
Database addition, deletion, modification and query of Python Sqlalchemy basic operation
2022-01-31 12:27:33 【Autumn in northern China】
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 :
- Simple and easy to read : Abstract a data table as an object ( Data model ), More visual and easy to read .
- portable : Encapsulate a variety of database engines , Dealing with multiple databases , The actual operation is basically the same , The code is easy to maintain .
- 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
copyright notice
author[Autumn in northern China],Please bring the original link to reprint, thank you.
https://en.pythonmana.com/2022/01/202201311227316104.html
The sidebar is recommended
- [algorithm learning] 1108 IP address invalidation (Java / C / C + + / Python / go / trust)
- Test platform series (71) Python timed task scheme
- Java AES / ECB / pkcs5padding encryption conversion Python 3
- Loguru: the ultimate Python log solution
- Blurring and anonymizing faces using OpenCV and python
- How fast Python sync and async execute
- Python interface automation test framework (basic) -- common data types list & set ()
- Python crawler actual combat, requests module, python realizes capturing video barrage comments of station B
- Python: several implementation methods of multi process
- Sword finger offer II 054 Sum of all values greater than or equal to nodes | 538 | 1038 (Java / C / C + + / Python / go / trust)
guess what you like
-
How IOS developers learn python programming 3-operator 2
-
How IOS developers learn python programming 2-operator 1
-
[Python applet] 8 lines of code to realize file de duplication
-
Python uses the pynvml tool to obtain the working status of GPU
-
Data mining: Python actual combat multi factor analysis
-
Manually compile opencv on MacOS and Linux and add it to Python / C + + / Java as a dependency
-
Use Python VTK to batch read 2D slices and display 3D models
-
Complete image cutting using Python version VTK
-
Python interface automation test framework (basic) -- common data types Dict
-
Django (make an epidemic data report)
Random recommended
- Python specific text extraction in actual combat challenges the first step of efficient office
- Daily python, Part 8 - if statement
- Django model class 1
- The same Python code draws many different cherry trees. Which one do you like?
- Python code reading (Chapter 54): Fibonacci sequence
- Django model class 2
- Python crawler Basics
- Mapping 3D model surface distances using Python VTK
- How to implement encrypted message signature and verification in Python -- HMAC
- leetcode 1945. Sum of Digits of String After Convert(python)
- leetcode 2062. Count Vowel Substrings of a String(python)
- Analysis of Matplotlib module of Python visualization
- Django permission management
- Python integrated programming -- visual hot search list and new epidemic situation map
- [Python data collection] scripy realizes picture download
- Python interface automation test framework (basic part) -- loop statement of process control for & while
- Daily python, Chapter 9, while loop
- Van * Python | save the crawled data with docx and PDF
- Five life saving Python tips
- Django frequency control
- Python - convert Matplotlib image to numpy Array or PIL Image
- Python and Java crawl personal blog information and export it to excel
- Using class decorators in Python
- Untested Python code is not far from crashing
- Python efficient derivation (8)
- Python requests Library
- leetcode 2047. Number of Valid Words in a Sentence(python)
- leetcode 2027. Minimum Moves to Convert String(python)
- How IOS developers learn Python Programming 5 - data types 2
- leetcode 1971. Find if Path Exists in Graph(python)
- leetcode 1984. Minimum Difference Between Highest and Lowest of K Scores(python)
- Python interface automation test framework (basic) -- basic syntax
- Detailed explanation of Python derivation
- Python reptile lesson 2-9 Chinese monster database. It is found that there is a classification of color (he) desire (Xie) monsters during operation
- A brief note on the method of creating Python virtual environment in Intranet Environment
- [worth collecting] for Python beginners, sort out the common errors of beginners + Python Mini applet! (code attached)
- [Python souvenir book] two people in one room have three meals and four seasons: 'how many years is it only XX years away from a hundred years of good marriage' ~?? Just come in and have a look.
- The unknown side of Python functions
- Python based interface automation test project, complete actual project, with source code sharing
- A python artifact handles automatic chart color matching