J'ai décidé de créer un processus qui prend en CSV de l'extérieur et le convertit en un fichier utilisé dans mon système.
Ce n'était pas si volumineux, donc ça aurait été bien d'écrire un programme procédural, mais j'avais du temps à perdre, alors j'ai sérieusement divisé les couches.
À ce moment-là, j'aimerais écrire comment je l'ai fait, ainsi que les exigences de l'échantillon.
Cette fois, y compris les études, je vais le faire avec Python 3.7 et Lambda.
https://github.com/jnuank/io_logic_isolation
Veuillez vous reporter à [Image de séparation finale](# Image de séparation finale).
Les données de transaction sont extraites de la caisse enregistreuse utilisée par une chaîne de magasins, et les transactions générées à chaque caisse enregistreuse de chaque magasin sont prises dans le CSV des ventes d'autres systèmes et converties en données qui peuvent être utilisées dans le propre système. À ce moment-là, le CSV lu cette fois ne se convertit pas en délimiteur d'espace tel quel, mais certains doivent être convertis à la valeur du système.
Les données pour chaque magasin et chaque caisse enregistreuse pour un certain jour sont rassemblées dans un fichier.
Données aspirées de la caisse enregistreuse
#Comment lire les modifications en fonction du troisième type d'enregistrement
# 01: 1:Code magasin 2:Numéro de caisse enregistreuse 3:Code de type(01:En-tête de transaction) 4:Numéro de transaction 5:YYYYMMDDHHMMSS
# 02: 1:Code magasin 2:Numéro de caisse enregistreuse 3:Code de type(02:Détails de la transaction) 4:Numéro de transaction 5:Nom du produit 6:Prix unitaire 7:Quantité
"1","1","01","0000001", "20200816100000"
"1","1","02","0000001","Produit A","1000", "2"
"1","1","02","0000001","Produit B","500", "4"
"1","1","02","0000001","Produit C","100", "10"
"1","1","01","0000002", "20200816113010"
"1","1","02","0000002","Produit D","10000", "1"
"1","1","02","0000002","Produit E","2000", "1"
"1","2","01","0000001", "20200816102049"
"1","2","02","0000001","Produit A","1000", "3"
"1","2","02","0000001","Produit D","10000", "2"
"1","2","02","0000001","Produit F","500", "5"
"1","2","02","0000001","Produit G","4400", "2"
"2","1","01","0000001", "20200816152009"
"2","1","02","0000001","Produit F","500", "1"
"2","1","02","0000001","Produit G","4400", "1"
Mois du magasin
# 1:Magasin code 2:Date de vente(YYYYMM) 3:Montant des ventes
001 202008 500000
002 202008 300000
003 202008 900000
Par jour de magasin
# 1:Magasin code 2:Date de la vente(YYYYMMDD) 3:Montant des ventes
001 20200816 51300
002 20200816 4900
Stocker les détails quotidiens
# 1:Magasin code 2:Numéro de transaction 3:Numéro de caisse enregistreuse 4:Temps de vente(YYYYMMDDHHMMSS) 5:Montant des ventes
001 0000001 001 20200816100000 5000
001 0000002 001 20200816113010 12000
001 0000001 901 20200816102049 34300
002 0000001 001 20200816152009 4900
―― Que voulez-vous savoir à partir des données téléchargées?
Je vais faire une telle Lambda.
Comme il ne s'agit que d'une conversion, il m'est facile d'écrire un solide logique dans la méthode du gestionnaire, mais j'ai tendance à le faire. Dans un tel cas, les événements suivants vous obligeront à changer la logique.
--Lorsque la structure des données CSV change --Lorsque la structure du fichier délimité par des espaces change ――Ce n'est pas CSV en premier lieu, ce n'est pas délimité par des espaces
Il semble que ** 2 types ** de demandes de changement soient possibles en raison de changements dans la structure des données d'entrée / sortie et des règles telles que les calculs.
En particulier lors de la réception de données d'un autre système comme cette fois, la définition de l'élément de données de l'autre partie peut être retardée ou il peut être difficile d'obtenir des données d'échantillon. Si le développement ne se déroulait pas là-bas, ou s'il était fait avec les spécifications que j'avais entendues verbalement, on dirait: «En fait, c'était une vieille information, donc c'est différent maintenant».
Personnellement, je n'étais pas habitué à Python, mais j'ai commencé à écrire des tests.
À ce stade, séparez le gestionnaire Lambda et la logique.
Le gestionnaire Lambda est celui qui reçoit une ** requête (événement, contexte) et renvoie finalement une réponse (état HTTP) **. C'est la responsabilité des gestionnaires de Lambda, alors gardez la logique de conversion hors des gestionnaires. Extraire les paramètres requis, les transférer dans la classe logique, recevoir les résultats et les inclure dans la réponse (si nécessaire).
handler.py
def import_handler(event, context):
try:
#Extraire les informations nécessaires de l'événement
body_str = event['body']
body = json.loads(body_str)
key = body['key']
bucket = os.environ['BUCKET_NAME']
#Importer CSV → Enregistrer séparés par des espaces
trans_app = CsvToSpaceSeparateApplication(bucket, key)
trans_app.csv_to_space_separate()
dict_value = {'message': 'téléversé', }
json_str = json.dumps(dict_value)
return {
'statusCode': 200,
'body': json_str
}
--Une fois que vous oubliez les ** entrées / sorties ** de Lambda, vous ne pouvez tester que la classe logique
En fait, je n'étais pas habitué à Python, donc j'ai pu écrire un test et essayer comment lire CSV et comment convertir en délimiteurs d'espace, donc c'était bien de séparer ici d'abord.
https://github.com/jnuank/io_logic_isolation/commit/539b7d8bcdf8ca1b253b6185ab88f0b98806f8b4
Je ne veux pas convertir le CSV lu tel quel en séparateurs d'espaces, mais je pense que certaines valeurs peuvent souhaiter être converties en valeurs du système.
Le code magasin et le numéro de caisse enregistreuse du système externe sont les numéros de série installés, À l'intérieur du système, il existe des différences dans le système de numérotation, comme le fait que chaque chiffre a une signification.
--Store code dans le système: 3 chiffres
Par conséquent, j'aimerais avoir une table pour la conversion, mais comme le magasin de données n'a pas été décidé, je pense qu'il y a des moments où j'utilise une table temporaire pour les tests. De plus, si vous écrivez les détails d'implémentation d'une source de données spécifique (comme l'établissement d'une connexion) dans la logique de conversion, vous devrez le modifier lorsque la source de données change.
Pour éviter cela et pour retarder la décision des détails d'implémentation, préparez une classe abstraite qui s'attend à renvoyer la valeur que le système a après avoir passé la valeur obtenue à partir de CSV pour le moment.
Classe abstraite pour la conversion des valeurs CSV en valeurs système
from abc import ABCMeta, abstractmethod
class CodeRepositoryBase(object, metaclass=ABCMeta):
"""
Classe abstraite pour obtenir le code de la banque de données
"""
@abstractmethod
def get_shop_code(self, external_system_shop_code: str) -> str:
"""
Obtenez le code du magasin
:param external_system_shop_code:Code de magasin numéroté par un système externe
:return:code magasin
"""
raise NotImplementedError()
@abstractmethod
def get_cash_register_code(self, external_system_shop_code: str, external_system_cash_register_code: str) -> str:
"""
Obtenez un numéro de caisse enregistreuse
:param external_system_shop_code:Code de magasin numéroté par un système externe
:param external_system_cash_register_code:Numéro de caisse enregistreuse attribué par un système externe
:return:Numéro de caisse enregistreuse
"""
raise NotImplementedError()
Référentiel de test avec des données dans dict
from source.domain.repository.code_repository_base import CodeRepositoryBase
class InMemoryCodeRepository(CodeRepositoryBase):
"""
Implémentation du référentiel en mémoire
"""
def __init__(self):
# key:Valeur du code de magasin du système externe:code magasin
self.__shop_code_table = {
'1': '001',
'2': '002',
'3': '003'
}
# key:(Code de magasin système externe,Numéro de caisse enregistreuse du système externe) value:Numéro de caisse enregistreuse
#Le premier chiffre du numéro d'enregistrement est "0":Caisse enregistreuse permanente, "9":Caisse enregistreuse d'événement
self.__cash_register_code_table = {
('1', '1'): '001',
('1', '2'): '901',
('2', '1'): '001',
}
def get_shop_code(self, external_system_shop_code: str) -> str:
"""
Obtenez le code du magasin
:param external_system_shop_code:Code de magasin numéroté par un système externe
:return:code magasin
"""
result = self.__shop_code_table.get(external_system_shop_code)
if result is None:
raise ValueError(f'Le code de magasin correspondant à la clé spécifiée n'existe pas. Clé:{external_system_shop_code}')
return result
def get_cash_register_code(self, external_system_shop_code: str, external_system_cash_register_code:str) -> str:
"""
Obtenez un numéro de caisse enregistreuse
:param external_system_shop_code:Code de magasin numéroté par un système externe
:param external_system_cash_register_code:Numéro de caisse enregistreuse attribué par un système externe
:return:Numéro de caisse enregistreuse
"""
result = self.__cash_register_code_table.get((external_system_shop_code, external_system_cash_register_code))
if result is None:
raise ValueError(f'Le numéro d'enregistrement correspondant à la clé spécifiée n'existe pas. Clé:{external_system_cash_register_code}')
return result
Code de test
from pytest import raises
from tests.In_memory_code_repository import InMemoryCodeRepository
class TestInMemoryCodeRepository:
def test_Le code de magasin 001 est renvoyé(self):
result = InMemoryCodeRepository().get_shop_code('1')
assert result == '001'
―― Depuis que la source de données a été décidée, cela a pris plus de temps que d'habitude car il était nécessaire de repenser les détails de l'implémentation.
https://github.com/jnuank/io_logic_isolation/commit/1c54107aafb72d3faee57b3ef85a5510f794deae
La séparation était possible jusqu'au point de convertir la valeur de CSV en valeur de propre système.
Maintenant, sur la base des données CSV suivantes, nous allons le convertir en une valeur séparée par des espaces.
[Repost] Données collectées à partir de la caisse enregistreuse
#Comment lire les modifications en fonction du troisième type d'enregistrement
# 01: 1:Code magasin 2:Numéro de caisse enregistreuse 3:Code de type(01:En-tête de transaction) 4:Numéro de transaction 5:YYYYMMDDHHMMSS
# 02: 1:Code magasin 2:Numéro de caisse enregistreuse 3:Code de type(02:Détails de la transaction) 4:Numéro de transaction 5:Nom du produit 6:Prix unitaire 7:Quantité
"1","1","01","0000001","20200816100000"
"1","1","02","0000001","Produit A","1000","2"
"1","1","02","0000001","Produit B","500","4"
"1","1","02","0000001","Produit C","100","10"
"1","1","01","0000002","20200816113010"
"1","1","02","0000002","Produit D","10000","1"
"1","1","02","0000002","Produit E","2000","1"
"1","2","01","0000001","20200816102049"
"1","2","02","0000001","Produit A","1000","3"
"1","2","02","0000001","Produit D","10000","2"
"1","2","02","0000001","Produit F","500","5"
"1","2","02","0000001","Produit G","4400","2"
"2","1","01","0000001","20200816152009"
"2","1","02","0000001","Produit F","500","1"
"2","1","02","0000001","Produit G","4400","1"
Écrivez le mappage basé sur le document d'élément de définition de table. Je l'ai écrit très grossièrement et ça ressemble à ça.
app.py
#Renvoie une liste mappée selon la définition de l'élément en fonction du CSV transmis
#Changer la liste en espace délimité du côté de l'appelant
@dataclass
class CsvToShopSales:
code_respository: CodeRepositoryBase
def csv_to_sales_by_shop(self, csv_list) -> List[List[str]]:
names_list = list(range(10))
df = pd.read_csv(csv_list, names=names_list, dtype='object').fillna('_')
SHOP_COLUMN = 0
#Regrouper par code magasin
shop_group_list = df.groupby(SHOP_COLUMN)
results = []
for group_rows in shop_group_list:
shop_code = self.code_respository.get_shop_code(group_rows[0])
year_month = [record[4] for record in group_rows[1].values.tolist() if record[2] == '01'][0][:6]
amount_list = [int(record[5]) * int(record[6]) for record in group_rows[1].values.tolist() if record[2] == '02']
sales_amount = sum(amount_list)
results.append([shop_code, year_month, str(sales_amount)])
return results
Que la structure des données de chaque côté des données avant la conversion ou les données après la conversion change, je pense qu'il n'est pas bon de modifier le même code. Je ne veux pas avoir plus d'une raison de changer pour une classe.
--Recevoir CSV et en faire un modèle qui extrait uniquement les données souhaitées en CSV (entrée / sortie) --Convertir du modèle CSV au modèle de domaine de vente (règle) --Convertir en données séparées par des espaces et enregistrer (entrée / sortie)
Séparons-les de app.py. Ci-dessous, une image du contenu de app.py séparé.
Référentiel pour convertir du CSV en modèle
from abc import ABCMeta, abstractmethod
from typing import List
from source.domain.models.csv_models.csv_cash_transaction_header import CsvCashTransactionHeader
class CsvCashTransactionRepositoryBase(object, metaclass=ABCMeta):
"""
Classe abstraite de référentiel pour recevoir les données de transaction de caisse enregistreuse CSV
"""
@abstractmethod
def load(self) -> List[CsvCashTransactionHeader]:
"""
Obtenir le modèle de données de transaction de caisse enregistreuse
:return:Modèle de données de transaction de caisse enregistreuse
"""
raise NotImplementedError()
@abstractmethod
def save(self, data: CsvCashTransactionHeader) -> None:
"""
Enregistrer le modèle de données de transaction de caisse enregistreuse
:param data:Modèle de données de transaction de caisse enregistreuse
"""
raise NotImplementedError('Je ne peux pas encore enregistrer')
Modèle CSV
from __future__ import annotations
from dataclasses import dataclass, field
from typing import List
from source.domain.models.csv_models.csv_cash_transaction_detail import CsvCashTransactionDetail
@dataclass(frozen=True, order=True)
class CsvCashTransactionHeader:
"""
Modèle CSV de données de transaction de caisse enregistreuse
"""
#code magasin
shop_code: str = field(compare=True)
#Numéro de caisse enregistreuse
cash_register_code: str = field(compare=True)
#Numéro de transition
transaction_code: str = field(compare=True)
#Temps de négociation
transaction_datetime: str = field(compare=True)
#Détails de la transaction
transaction_details: List[CsvCashTransactionDetail]
from dataclasses import dataclass
@dataclass(frozen=True)
class CsvCashTransactionDetail:
"""
Modèle CSV de détails des données de transaction de caisse enregistreuse
"""
#Nom du produit
item_name: str
#Prix unitaire
unit_price: int
#quantité
quantity: int
En tant que modèle, cela ressemble à ceci.
Créez un modèle de domaine comme décrit dans [Image du modèle de données après importation](# Image du modèle de données après importation).
from dataclasses import dataclass, field
from functools import reduce
from operator import add
from typing import List
from source.domain.models.salses.daily_sales import DailySales
@dataclass(frozen=True)
class ShopMonthlySales:
shop_code: str
year_month: str
daily_sales_list: List[DailySales] = field(default_factory=list, compare=False)
def amount(self) -> int:
return reduce(add, map(lambda data: data.amount(), self.daily_sales_list))
from dataclasses import dataclass, field
from datetime import datetime
from functools import reduce
from operator import add
from typing import List
from source.domain.models.salses.daily_sales_detail import DailySalesDetail
@dataclass(frozen=True)
class DailySales:
sales_date: datetime.date
details: List[DailySalesDetail] = field(default_factory=list, compare=False)
def amount(self) -> int:
return reduce(add, map(lambda data: data.amount, self.details))
import datetime
from dataclasses import dataclass
@dataclass(frozen=True)
class DailySalesDetail:
transaction_code: str
transaction_datetime: datetime.datetime
cash_number: str
amount: int
Créez une classe de règles qui prend un modèle CSV et le transforme en modèle de domaine de vente.
@dataclass(frozen=True)
class TransferRules(object):
"""
Classe de règles de conversion
"""
repository: CodeRepositoryBase
def to_shop_sales(self, sources: List[CsvCashTransactionHeader]) -> List[ShopMonthlySales]:
results: List[ShopMonthlySales] = []
sources.sort(key=lambda x: x.shop_code)
#Regrouper par magasin et convertir en modèle
for key, g in groupby(sources, key=lambda x: x.shop_code):
shop_code = self.repository.get_shop_code(key)
details: List[DailySalesDetail] = []
dt = ''
day = ''
year_month = ''
for member in g:
dt = datetime.strptime(member.transaction_datetime, '%Y%m%d%H%M%S')
day = date(dt.year, dt.month, dt.day)
year_month = member.transaction_datetime[:6]
cash_register_code = self.repository.get_cash_register_code(member.shop_code, member.cash_register_code)
amount = sum([s.unit_price * s.quantity for s in member.transaction_details])
detail = DailySalesDetail(member.transaction_code,
dt,
cash_register_code,
amount)
details.append(detail)
daily = DailySales(day, details)
shop_sales = ShopMonthlySales(shop_code, year_month, [daily])
results.append(shop_sales)
return results
Créez une classe qui stocke le modèle de domaine dans la banque de données, séparés par des espaces.
Cette fois, nous allons créer une classe à enregistrer dans S3.
class S3ShopSalesRepository(ShopSalesRepositoryBase):
"""
Implémentation d'un référentiel de vente en mémoire
"""
__bucket_name: str
def __init__(self, bucket_name):
self.__bucket_name = bucket_name
def save(self, sources: List[ShopMonthlySales]) -> None:
self.shop_monthly_sales = []
self.daily_sales = []
self.daily_details = []
for source in sources:
self.shop_monthly_sales.append(
[source.shop_code, source.year_month, str(source.amount())]
)
for daily in source.daily_sales_list:
self.daily_sales.append([
source.shop_code,
daily.sales_date.strftime('%Y%m%d'),
str(daily.amount()),
])
for detail in daily.details:
self.daily_details.append(
[source.shop_code,
detail.transaction_code,
detail.cash_number,
detail.transaction_datetime.strftime('%Y%m%d%H%M%S'),
str(detail.amount)]
)
self.shop_monthly_sales = self.__comma2dlist_to_space2dlist(self.shop_monthly_sales)
self.daily_sales = self.__comma2dlist_to_space2dlist(self.daily_sales)
self.daily_details = self.__comma2dlist_to_space2dlist(self.daily_details)
try:
self.__s3_upload(self.shop_monthly_sales, self.__bucket_name, 'Ventes en magasin.txt')
self.__s3_upload(self.daily_details, self.__bucket_name, 'Par jour de magasin.txt')
self.__s3_upload(self.daily_details, self.__bucket_name, 'Stocker les détails quotidiens.txt')
except Exception as error:
raise error
Le processus d'enregistrement dans le magasin de données et le processus de conversion en délimiteurs d'espace sont combinés dans cette classe.
La raison en est que cette fois, il est converti en espace délimité, mais lors de la conversion vers un autre magasin de données, il peut être converti dans un autre format, donc dans la classe d'implémentation de ShopSalesRepositoryBase
Je vous laisse ça.
handler.py
def import_handler(event, context):
try:
#Extraire les informations nécessaires de l'événement
body_str = event['body']
body = json.loads(body_str)
key = body['key']
bucket_name = os.environ['BUCKET_NAME']
code_repository = InMemoryCodeRepository()
csv_repository = S3CsvCashTransactionRepository(key, bucket_name)
#En supposant que le seau a déjà été décidé
shop_sales_repository = S3ShopSalesRepository('xxxxx-bucket')
#Importer CSV → Enregistrer séparés par des espaces
trans_app = CsvToSpaceSeparateApplication(code_repository, csv_repository, shop_sales_repository)
trans_app.csv_to_space_separate()
#Assemblage de réponse
dict_value = {'message': 'téléversé', }
json_str = json.dumps(dict_value)
return {
'statusCode': 200,
'body': json_str
}
except ValueError as error:
logger.exception(f'{error}')
dict_value = {'message': f'{error}', }
json_str = json.dumps(dict_value)
return {
'statusCode': 500,
'body': json_str
}
except Exception as error:
logger.exception(f'{error}')
dict_value = {'message': f'Une erreur de traitement s'est produite. Veuillez réessayer après un certain temps', }
json_str = json.dumps(dict_value)
return {
'statusCode': 500,
'body': json_str
}
application
@dataclass
class CsvToSpaceSeparateApplication(object):
"""
CSV → Processus de conversion délimité par un espace
"""
code_repository: CodeRepositoryBase
csv_repository: CsvCashTransactionRepositoryBase
shop_sales_repository: ShopSalesRepositoryBase
def csv_to_space_separate(self) -> None:
"""
CSV → conversion délimitée par un espace
"""
#Convertir en modèle CSV
csv_models = self.csv_repository.load()
#Convertir en modèle de domaine
shop_monthly_sales = TransferRules(self.code_repository).to_shop_sales(csv_models)
#Convertir en espace délimité et enregistrer
self.shop_sales_repository.save(shop_monthly_sales)
L'image de CsvToSpaceSeparateApplication appelée par le gestionnaire ressemble à ceci. Chaque procédure "sortie-> conversion-> sortie" est exprimée par la méthode dans la couche Application.
L'intention de chaque processus est également exprimée en le regroupant dans une classe.
https://github.com/jnuank/io_logic_isolation/commit/b4e8885b2f269a608d0cfe3bfb414d4135277022
J'ai essayé de pratiquer une configuration similaire sur le terrain,
―― Étant donné que l'entrée / sortie et la conversion ont été séparées et qu'elle est sur le point d'être libérée, j'ai été informé que la configuration CSV des autres systèmes va changer.
Bien que ce fût un petit processus, j'ai essayé cette fois de séparer les règles d'entrée / sortie et de calcul / jugement, et j'ai estimé que cela serait utile pour construire un grand système à l'avenir. Il semble qu'il y aura des discussions sur la rentabilité, mais si j'ai un peu de temps, j'aimerais essayer d'en être informé régulièrement. (Bien sûr, ce n'est pas le cas avec le code que vous prévoyez de jeter, mais la plupart du temps, ce n'est pas le cas ...)