Location>code7788 >text

When implementing data processing in SQLAlchemy, there are some common experiences in implementing table self-referencing, many-to-many, union queries, ordered ids, etc.

Popularity：292 ℃/2024-08-26 13:48:12

Sometimes, when we use SQLAlchemy to manipulate certain tables, we need to use foreign key relationships to realize one-to-many or many-to-many relational references, as well as joint queries on multiple tables, with sequential uuid values or self-incremented id values, string splitting, and other common processing operations.

1. Define in SQLAlchemy with nestedchildren Table of relations

To define a SQLAlchemy with a nestedchildren relationship, such as a table containingid cap (a poem)pid field, you can use therelationship cap (a poem)ForeignKey to establish a father-son relationship.

First, you need to define a model class that contains theid cap (a poem)pid Fields.id is the primary key.pid is a foreign key that points to the parent record. Then, you use therelationship to establish a father-son relationship.

from sqlalchemy import Column, Integer, String, ForeignKey
from  import relationship, declarative_base
from  import create_async_engine, AsyncSession
from  import sessionmaker

Base = declarative_base()

class DictTypeInfo(Base):
    __tablename__ = 'dict_type_info'
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    code = Column(String)
    remark = Column(String)
    seq = Column(Integer)
    pid = Column(Integer, ForeignKey('dict_type_info.id'))  # The foreign key points to the id of the parent node

    # Define parent relationship
    parent = relationship("DictTypeInfo", remote_side=[id], back_populates="children")

    # Define children relationship
    children = relationship("DictTypeInfo", back_populates="parent")

The example usage code is shown below.

# Creating Asynchronous Engines and Sessions
DATABASE_URL = "mysql+asyncmy://username:password@localhost/mydatabase"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)

async def init_db():
    async with () as conn:
        await conn.run_sync(.create_all)

# Example: How to insert data and query
async def example_usage():
    async with AsyncSessionLocal() as session:
        async with ():
            # insert data
            parent_node = DictTypeInfo(name="Parent", code="P001", remark="Parent Node", seq=1)
            child_node1 = DictTypeInfo(name="Child1", code="C001", remark="First Child", seq=1, parent=parent_node)
            child_node2 = DictTypeInfo(name="Child2", code="C002", remark="Second Child", seq=2, parent=parent_node)
            (parent_node)
            (child_node1)
            (child_node2)
            
        # Query Data
        async with ():
            result = await (
                "SELECT * FROM dict_type_info WHERE pid IS NULL"
            )
            parent_nodes = ().all()
            for node in parent_nodes:
                print(f"Parent Node: {}, Children: {[ for child in ]}")

Code Description

  1. Defining Model Classes (DictTypeInfo):

    • id: Primary key.
    • pid: A foreign key that points to the same table'sid, denotes the parent node.
    • parent: Parentage throughremote_side Sets the foreign key of this model to point to its own primary key.
    • children:: Sub-relationships.back_populates For mapping of bi-directional relationships.

  2. Creating Asynchronous Engines and Sessions:

    • utilizationcreate_async_engine cap (a poem)AsyncSession Create database engines and sessions to support asynchronous operations.

  3. Inserting and querying data:

    • The Insert Data example shows how to create parent and child nodes and associate child nodes to the parent.
    • The Query Data example shows how to query all parent nodes and their children.

caveat

  • remote_side: Inrelationship Middle.remote_side is to specify which fields are on the remote side (i.e., the target of the child node relationship).
  • Ensure that the correct foreign key constraints are defined in the model. In the model you provided, thepid column needs to point to the same table in theid Column. EnsureForeignKey Set up correctly.
  • asynchronous operation: UseAsyncSession cap (a poem)asyncio Perform asynchronous database operations.
  • Create Table: Ensure that the table structure is correct when initializing the database.

To useselectinload Load apid objects and their sublists can be accessed through SQLAlchemy'sselectinload to optimize the loading subrelationships.selectinload You can reduce the number of SQL queries, especially when loading data with a hierarchical structure.

async def get_tree(pid: int):
    async with AsyncSessionLocal() as session:
        # Load all child nodes via selectinload
        stmt = select(DictTypeInfo).filter( == pid).options(selectinload())
        result = await (stmt)
        nodes = ().all()
        
        return nodes

In this way, a call toget_tree function to get the specifiedpid of the node and its children, the code is as follows.

async def example_usage():
    nodes = await get_tree(pid=1)
    for node in nodes:
        print(f"Node: {}, Children: {[ for child in ]}")

selectinload: selectinload The N+1 query problem can be reduced by loading the relevant objects with an additional query. This is suitable for loading hierarchical data. In this way, you can use SQLAlchemy'sselectinload to efficiently load objects with parent-child relationships and optimize database query performance.

 

Similarly, our implementation of many-to-many relationships in SQLAlchemy is handled in a similar way.

Implementing a many-to-many relationship in SQLAlchemy usually involves creating an association table that will store the foreign keys of the two associated tables, thus implementing a many-to-many relationship. The following is a detailed step-by-step procedure to implement a many-to-many relationship.

1) Define an association table for many-to-many relationships

First, a related table needs to be defined that contains two foreign keys that point to the primary table at each end. This is usually done using theTable object to implement it.

from sqlalchemy import Table, Column, Integer, ForeignKey
from  import declarative_base

Base = declarative_base()

association_table = Table('association', ,
    Column('left_id', Integer, ForeignKey('left_table.id')),
    Column('right_id', Integer, ForeignKey('right_table.id'))
)

In this example, theassociation_table is an intermediate table containing two foreign keys:left_id cap (a poem)right_id point respectivelyleft_table cap (a poem)right_table The primary key of the

2) Define the model at both ends and add the relationship

At both ends of the model, use therelationship to define a many-to-many relationship and specify thesecondary The parameter is the association table.

from  import relationship

class LeftModel(Base):
    __tablename__ = 'left_table'

    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    rights = relationship("RightModel", secondary=association_table, back_populates="lefts")

class RightModel(Base):
    __tablename__ = 'right_table'

    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    lefts = relationship("LeftModel", secondary=association_table, back_populates="rights")
  • rights beLeftModel The relational attribute defined in theRightModel。
  • lefts beRightModel The relational attribute defined in theLeftModel。
  • secondary=association_table Tell SQLAlchemy to useassociation_table as a join table.
  • back_populates Symmetric references for bidirectional relations.

3) Create database and insert data

The following code shows how to create a database, insert data and query a many-to-many relationship.

from sqlalchemy import create_engine
from  import sessionmaker

# Creating a Database Engine
engine = create_engine('sqlite:///')
.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

# Creating model instances
left1 = LeftModel(name="Left 1")
right1 = RightModel(name="Right 1")
right2 = RightModel(name="Right 2")

# Setting up a many-to-many relationship
 = [right1, right2]

# Add to session and submit
(left1)
()

# Query and print relationships
for right in :
    print()  # Output: Right 1, Right 2

for left in :
    print()  # Output: Left 1

You can manipulate these relationships as you would a normal list, such as adding, removing associations, and so on:

# Add Relationship
(RightModel(name="Right 3"))
()

# Delete relationship
(right2)
()

With these steps, you can implement and manipulate many-to-many relationships in SQLAlchemy.

 

2、Record association query in SQLAlchemy by joining multiple tables.

For example, in my framework, dictionary categories and dictionary items are managed in different tables, so if you need to do a query for a dictionary item based on the category name, then you need to join the two tables for processing.

This is done by creating a query that willDictDataInfo traveling allowanceDictTypeInfo Table joins (viaDictType_ID cap (a poem)Id (columns)

from  import select
from  import aliased
from  import AsyncSession
from  import create_async_engine
from  import sessionmaker

# Assuming your database models are DictDataInfo and DictTypeInfo
# The two model classes need to be defined in advance

DATABASE_URL = "mysql+asyncmy://username:password@localhost/mydatabase"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)

async def get_dict_data(dict_type_name: str):
    async with AsyncSessionLocal() as session:
        # Creating an Alias
        DictData = aliased(DictDataInfo)
        DictType = aliased(DictTypeInfo)

        # Federated queries and filtering based on conditions
        stmt = (
            select(DictData)
            .join(DictType, DictData.DictType_ID == )
            .filter( == dict_type_name)
        )

        result = await (stmt)
        dict_data = ().all()

        return dict_data

# Sample usage
import asyncio

async def example_usage():
    dict_type_name = "some_type_name"
    dict_data = await get_dict_data(dict_type_name)
    for data in dict_data:
        print(data)

Code Description

  1. aliased: Usealiased Create aliases for tables so that they can be easily referenced in queries.

  2. join: Usejoin Perform a table join. HereDictDataInfo homogeneousDictType_ID columnsDictTypeInfo homogeneousid Column Connections.

  3. filter: Usefilter to add a conditional filter that filters outDictTypeInfo statistical tablesname Columns are equal todict_type_name of the record.

  4. select: Useselect statement to select theDictDataInfo table, which corresponds to the records in theSelect(d => d)。

  5. Asynchronous operations: Since SQLAlchemy's asynchronous mode is used, all database operations are performed in theasync with cap (a poem)await statement is performed to ensure asynchronous execution.

If we need to convert the obtained data into an object, we can use the following processing code to achieve this.

# Define the CListItem class
class CListItem:
    def __init__(self, name, value):
         = name
         = value

# Defining example lists and conversion operations
def convert_list_items(list_items):
    dict_list = []
    if list_items:  # Make sure list_items is not None
        for info in list_items.Items:
            dict_list.append(CListItem(, ))
    return dict_list

 

3, the use of sqlalchemy insert data, how to determine the non-self-augmenting type of time, id assignment of a sequence of uuid value

Sometimes, our data table primary key is in string, this applies to a wide range of uses, it is easier to determine the value of the id key at the time of insertion, so that you can process the relevant content.

However, sometimes we can have the backend do the work of determining an ordered ID value, so using SQLAlchemy how should we accomplish this?

First, make sure you have imported theuuid library, the standard Python library for generating UUIDs.

Ordered UUIDs are usually time-based UUIDs. you can use theuuid.uuid1() to generate a time-based UUID.

def generate_sequential_uuid():
    return uuid.uuid1()  # Generate ordered UUIDs based on time

When defining a SQLAlchemy model, you can set theid field is set to the UUID that was generated using this function. this is usually done in the model via thedefault The parameter sets the default value.

from sqlalchemy import Column, String
from  import declarative_base

Base = declarative_base()

class MyModel(Base):
    __tablename__ = 'my_table'

    id = Column(String(36), primary_key=True, default=generate_sequential_uuid, nullable=False)
    # Other fields ...

When inserting new data, if theid field is empty, it will automatically use thegenerate_sequential_uuid function generates a time-based UUID.

This ensures that when inserting data, non-self-incrementing types ofid field is assigned a sequential UUID value.

For self-incrementing integersidSQLAlchemy provides an automatic processing mechanism. All you need to do is set theid The field is defined asInteger type and set theprimary_key=TrueSQLAlchemy automatically sets the self-incrementing attribute for the field.

from sqlalchemy import Column, Integer, String
from  import declarative_base

Base = declarative_base()

class MyModel(Base):
    __tablename__ = 'my_table'

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50))
    # Other fields...

By default, SQLAlchemy uses the database's native self-increment mechanism (e.g., MySQL'sAUTO_INCREMENT or PostgreSQL'sSERIAL). If you need to use a customized self-incrementing policy, you can do so by setting theSequence to implement (for those who supportSequence database, such as PostgreSQL).

from sqlalchemy import Sequence

class MyModel(Base):
    __tablename__ = 'my_table'

    id = Column(Integer, Sequence('my_sequence'), primary_key=True)
    name = Column(String(50))

In the above code, theSequence('my_sequence') defines a sequence that SQLAlchemy will use to generate a self-incrementingid Value.

With these steps, you can easily deal with integrative self-enhancementid field, SQLAlchemy automatically assigns each new record a unique self-incrementingid。

 

4. Data processing of strings when inserting records

When inserting data dictionary in batch, I would like to transform it based on the user input content (multiple rows of data), split each row for judgment, and process the insertion if it meets the conditions.

In Python, you can use the string'ssplitlines() method to achieve the same functionality.

# Assuming Data and are taken from the input
Data = "example\nline1\nline2\n"  # sample data
input_seq = "123"  # Example Sequence String

# Split Data by Row and Remove Empty Rows
array_items = [line for line in () if line]

# Initializing Variables
int_seq = -1
seq_length = 3
str_seq = input_seq

# Trying to convert str_seq to an integer
if str_seq.isdigit():
    int_seq = int(str_seq)
    seq_length = len(str_seq)

# Print results
print(f"Array Items: {array_items}")
print(f"int_seq: {int_seq}")
print(f"seq_length: {seq_length}")
  • Pythonsplitlines() method splits the string on a line-by-line basis, and automatically handles various line breaks (including the\n cap (a poem)\r\n）。
  • list-deductive formula[line for line in () if line] removes blank lines, similar to C#'s。
  • utilizationstr_seq.isdigit() probestr_seq is composed entirely of numbers, similar to C#'s。

In Python, you can use the() function to split a string according to a regular expression. Here is the corresponding Python code:

import re

# Suppose info is an object with Name and Value attributes.
class Info:
    def __init__(self):
         = ""
         = ""

info = Info()

# dictData is the input string
dict_data = "example_name example_value"

# Splitting strings by whitespace using regular expressions
array = (r'\s+', dict_data)

# Attributes assigned to the info object
 = array[0]
 = array[1] if len(array) > 1 else array[0]

# Print results
print(f"Name: {}")
print(f"Value: {}")

utilization() function to split the string according to whitespace characters (including spaces, tabs, etc.)dict_data。r'\s+' is a regular expression that represents one or more whitespace characters.

If you need to split a string based on multiple delimiters, you can also use a regular expression (re module) of the() Methods.

str_item = " 1,2,3;4;5/6/7、8、9；10 "

import re

result = (r"[;,|/，；、]+", str_item.strip())
print(result)

Result Output：['1', '2', '3', '4', '5', '6', '7', '8', '9', '10']

Explanation:

  • r'[[;,|/,;,]', text) in (r'[;,|/,;,]' is a regular expression pattern:
    [] Indicates a character class, meaning that it matches any character in the character class.
    ;,|/,;, denote semicolon, comma, vertical line, Chinese comma, Chinese semicolon, and space, respectively, which will be used as separators.

The use of regular expressions provides the flexibility to handle multiple separators for more complex segmentation needs.

 

<Previous Next>