This is the 15th day of my participation in the November Gwen Challenge. Check out the event details: The last Gwen Challenge 2021
Hello ~ I’m Milo! I am building an open source interface testing platform from 0 to 1, and I am also writing a complete tutorial corresponding to it. I hope you can support it. Welcome to pay attention to my public number test development pit goods, get the latest article tutorial!
review
In the last section we introduced the idea of optimizing the Dao logic, so today we will try to implement it and apply it to the development of Redis configuration management.
Think about the list method
Py creates a new class, Mapper, from which all dao classes will inherit.
Think about what a list needs, generally, fields, parameters, whether it’s like or equal to these three important things.
With this in mind, our pseudocode is ready to write:
# 1. Get session
async with async_session() as session:
condition = [model.deleted_at == 0]
# 2. Construct the query condition based on the field value in the parameterDatabasehelper. where(field, field value, condition)# 3. Query the result, the follow-up is consistent with the original way, do not write
Copy the codeAs you can see, in addition to the above three information, you also need to try to write to the log, that is, you need to create the log member variable, and you need the model class, otherwise you don’t know what table to change.
thinking
If value is not empty, then we need to query according to this condition.
Model and log, we can pass from subclass to parent class (setattr and getattr are used here) through class decorator.
class Mapper(object) :
log = None
model = None
@classmethod
async def list_data(cls, **kwargs) :
try:
async with async_session() as session:
SQL > select * from 'delete'
condition = [getattr(cls.model, "deleted_at") = =0]
Pass if the argument is not None
for k, v in kwargs.items():
# check if it is likeTODO:The in query is not supported here
like = v is not None and len(v) > 2 and v.startswith("%") and v.endswith("%")
# if the mode is like, use model. otherwise, use Model. Field is equal to the
DatabaseHelper.where(v, getattr(cls.model, k).like(v) if like else getattr(cls.model, k) == v,
condition)
result = await session.execute(select(cls.model).where(*condition))
return result.scalars().all(a)except Exception as e:
Call CLS log, write log and throw exception
cls.log.error(F "get{cls.model}List failed, error:{e}")
raise Exception(F "Failed to obtain data")
Copy the codeSince the field is a variable, we can’t use model. Field, so getattr(Model, field) instead. For those unfamiliar, check out Getattr.
So, we have a rough list method, but this one doesn’t have pagination, so we need to add a pagination mode, which is really easy.
@classmethod
async def list_data_with_pagination(cls, page, size, **kwargs) :
try:
async with async_session() as session:
condition = [getattr(cls.model, "deleted_at") = =0]
for k, v in kwargs.items():
like = v is not None and len(v) > 2 and v.startswith("%") and v.endswith("%")
sql = DatabaseHelper.where(v, getattr(cls.model, k).like(v) if like else getattr(cls.model, k) == v,
condition)
return await DatabaseHelper.pagination(page, size, session, sql)
except Exception as e:
cls.log.error(F "get{cls.model}List failed, error:{e}")
raise Exception(F "Failed to obtain data")
Copy the codeIt’s pretty much the same thing, except we go back there and call pagination.
Take a look at the DAO decorator
We’re bad enough to put this decorator on a subclass and pass model and log to the parent class. After all, methods are calling methods of the parent class, and the parent class can’t get data directly from the child class.
This avoids the awkward situation of having to pass in model and log when calling the list method.
Refine other methods
After the list is done, can the others be far behind? But it really seems to be a little bit of a problem, because we usually have some sort of repeat judgment, but that’s okay, we can write a query method.
@classmethod
def query_wrapper(cls, **kwargs) :
condition = [getattr(cls.model, "deleted_at") = =0]
Pass if the argument is not None
for k, v in kwargs.items():
# check if it is likeTODO:The in query is not supported here
like = v is not None and len(v) > 2 and v.startswith("%") and v.endswith("%")
# if the mode is like, use model. otherwise, use Model. Field is equal to the
DatabaseHelper.where(v, getattr(cls.model, k).like(v) if like else getattr(cls.model, k) == v,
condition)
return select(cls.model).where(*condition)
@classmethod
async def query_record(cls, **kwargs) :
try:
async with async_session() as session:
sql = cls.query_wrapper(**kwargs)
result = await session.execute(sql)
return result.scalars().first()
except Exception as e:
cls.log.error(F "query{cls.model}Failure, the error:{e}")
raise Exception(F "Failed to query data")
Copy the codeThe query_wrapper method was chosen because the query method was too generic.
- Write the insert interface normally
@classmethod
async def insert_record(cls, model) :
try:
async with async_session() as session:
async with session.begin():
session.add(model)
await session.flush()
session.expunge(model)
return model
except Exception as e:
cls.log.error(F "add{cls.model}Failed to record, error:{e}")
raise Exception(F "Failed to add record")
Copy the codeSo I’m returning model, if I don’t need it, I don’t have to, but I’m still returning it.
- Insert the interface layer
Same as before, check first, if not shut down again.
But this creates two sessions, open > Closed > Open > Closed
But it also decouples the query and insert operations.
- Write delete and modify methods
@classmethod
async def update_record_by_id(cls, user, model, not_null=False) :
try:
async with async_session() as session:
async with session.begin():
query = cls.query_wrapper(id=model.id)
result = await session.execute(query)
original = result.scalars().first()
if original is None:
raise Exception("Records do not exist.")
DatabaseHelper.update_model(original, model, user, not_null)
await session.flush()
session.expunge(original)
return original
except Exception as e:
cls.log.error(F "update{cls.model}Failed to record, error:{e}")
raise Exception(F "Failed to update records")
@classmethod
async def delete_record_by_id(cls, user, id) :
Delete :param user: :param id: :return: ""
try:
async with async_session() as session:
async with session.begin():
query = cls.query_wrapper(id=id)
result = await session.execute(query)
original = result.scalars().first()
if original is None:
raise Exception("Records do not exist.")
DatabaseHelper.delete_model(original, user)
except Exception as e:
cls.log.error(F "delete{cls.model}Failed to record, error:{e}")
raise Exception(F "Failed to delete record")
@classmethod
async def delete_by_id(cls, id) :
""" Physical delete :param ID: :return: ""
try:
async with async_session() as session:
async with session.begin():
query = cls.query_wrapper(id=id)
result = await session.execute(query)
original = result.scalars().first()
if original is None:
raise Exception("Records do not exist.")
session.delete(original)
except Exception as e:
cls.log.error(F "logical deletion{cls.model}Failed to record, error:{e}")
raise Exception(F "Failed to delete record")
Copy the codeDelete side supports physical delete and logical delete, of course, we usually use soft delete.
Improve other interfaces
You can see that the deletion and modification are the same as before. These steps are done. Redis management work can be smoothly, then we need to write pages for it!
Since this is a very basic table page, we won’t go over it. In the next section, we’ll write the redisManager using configured Redis connections to manage our connection data, setting the stage for executing Redis online and using it as a pre-condition.