Dans l'article précédent, j'ai essayé d'obtenir rapidement des données d'AS / 400 en utilisant pypyodbc, vous pouvez maintenant établir une connexion ODBC avec IBM i. Donc, j'ai trouvé beaucoup de choses que je voulais faire, mais quand j'y pensais, je devais le préparer, alors j'ai créé cet espace cette fois.
La raison pour laquelle il est "n ° 1" est qu'il est toujours au milieu de la route et que seule la partie d'entrée peut être traitée. La sortie est difficile à vérifier sans la machine réelle, donc cette fois elle est passée.
Je pense que je vais l'utiliser plus tard, donc je vais emballer diverses choses dans un fichier (module) appelé "sql_helper.py". C'est une fonctionnalité de base que vous pourriez trouver si vous recherchez un package, mais j'ai réinventé la roue en tant que révision de Python.
Cela ressemble à ça pour le moment.
sql_helper.py
def _load_lines(filename, encoding='utf-8'):
return [ l for l in open(filename, 'r', -1, encoding) ]
def load_raw_string(filename, encoding='utf-8', rstrip=True, lstrip=False):
def _echo(x): return x
if ( rstrip and lstrip): func = str.strip
elif (not rstrip and lstrip): func = str.lstrip
elif ( rstrip and not lstrip): func = str.rstrip # default
else: func = _echo # dummy
if rstrip: term = "\n" # cause rstrip(None) is remove "\n"!
else: term = ""
s = _load_lines(filename, encoding)
return term.join( list(map(lambda x: func(x), s)) )
def load_sql(filename, encoding='utf-8', lstrip=False):
sts = []
for st in strip_comment( load_raw_string(filename, encoding, True, lstrip) ).split(";\n"):
if (st.isspace() or st == "" ): continue
sts.append(st.strip("; \t\n") )
return sts
def load_config(filename, encoding='utf-8'):
cf = {}
for line in strip_comment(load_raw_string(filename, encoding)).split("\n"):
if ( line == "" or line.isspace() ): continue
key_, val_ = line.split("=")
cf[ key_.strip() ] = val_.strip()
return cf
# ...Continue encore un peu...
Je vais vous expliquer brièvement.
_load_lines()
Lisez le fichier spécifié. Faites simplement une liste de 1 ligne, 1 élément et vous avez terminé. Je ne veux tout simplement pas écrire «open (xxx, ...). Readline ()» «à chaque fois.
load_raw_string() La liste obtenue en utilisant `` _load_line () '' ci-dessus est lue, multipliée par lstrip () et rstrip () pour chaque ligne, puis reconstruite en une seule chaîne de caractères et renvoyée. Dois-je être en mesure de choisir de supprimer ou non le blanc au début de la ligne et le blanc à la fin de la ligne? Après avoir pris une décision facile, le code est devenu plus gros. Avant _load_lines () est la fonctionnalité que nous avons ajoutée pour cela. (Il est également dit que je veux terminer l'instruction de retour sur une ligne)
func dépend du paramètre, str.lstrip、str.rstrip、str.strip、_echo(Renvoyez simplement la valeur d'entrée)Est en cours de remplacement.
Si vous appelez str.rstrip et str.strip sans arguments, les sauts de ligne à la fin de la ligne disparaîtront également, donc si vous spécifiez supprimer la fin de la ligne, mettez un symbole de saut de ligne dans `` terme '', sinon un caractère vide. Est inséré et reconstruit avec une instruction return.
load_sql()
Obtenez le SQL à partir du nom de fichier. Si ";" (+ saut de ligne) est inclus dans ceci, divisez-le là et renvoyez-le à la liste des instructions SQL et vous avez terminé.
C'était bien jusqu'à ce que je réfléchisse, mais pour réaliser cela, j'ai fini par faire des aides pour les aides.
Par exemple, ce serait mauvais s'il était coupé par "; \ n" dans le commentaire, non? Ou, ce serait mauvais s'il était coupé par le "; \ n" dans la citation, non? Ou quelque chose.
En tant que cas composé, si c'est `` `` ...; / * block comment * / ``, il ne correspondra pas à "; \ n" si vous devez à nouveau supprimer le blanc en fin de ligne après avoir supprimé le commentaire! Etc....
Ce que je fais enfin est comme ça
1. Supprimez les blancs de fin de ligne en utilisant load_row_string ()
2. Supprimez le commentaire avec strip_comment () (décrit plus loin), puis supprimez le blanc à la fin de la ligne suite à la suppression du commentaire.
3. Enfin, divisez en une seule instruction SQL avec "; \ n"
4. Supprimez le ";" lui-même à la fin de chaque phrase
5. Remballez 4. et retournez
Il aurait peut-être été plus rapide de créer normalement un simple analyseur SQL.
load_config()
Lorsque vous lisez un fichier au format suivant ...
#### **`sample_connection.txt`**
```text
/* supports sql style comments */
driver = {iSeries Access ODBC Driver}
system = 127.0.0.1
uid = USER01
pwd = USER01
DefaultLibraries = MYLIB,TESTLIB1,TESTLIB2library
Convertissez en un dictionnaire comme celui-ci.
{'DefaultLibraries': 'MYLIB,TESTLIB1,TESTLIB2',
'driver': '{iSeries Access ODBC Driver}',
'pwd': 'USER01',
'system': '127.0.0.1',
'uid': 'USER01'}
Si vous passez ce dictionnaire à la fonction pypyodbc.connect (), vous n'avez pas à écrire de paramètres dans une ligne.
odbctest.py
from pypyodbc import connect
import sql_helper as helper
config = helper.load_config( "sample_connection.txt" )
with connect( **config ) as connection:
#Traitement d'accès à la base de données...
strip_comment() Au fur et à mesure que la source s'allonge, elle devient une fonction de suppression de commentaire qui a été utilisée.
sql_helper.py
from enum import Enum, auto
class States(Enum):
Statement = auto()
Quoted = auto()
Comment = auto()
End = auto()
_LimitLen = 999999999
def _singlequote_begin(s, b):
p = s.find("'", b)
return p if (p >= 0) else _LimitLen
def _singlequote_end(s, b):
p = s.find("'", b)
if (p >= 0 and p == s.find("''", b)): p = _singlequote_end(s, p + 2) # find recursive
return (_next, p, 0, States.Statement) if (p >= 0) else (None. _LimitLen, 0, States.End)
def _doublequote_begin(s, b):
p = s.find('"', b)
return p if (p >= 0) else _LimitLen
def _doublequote_end(s, b):
p = s.find('"', b)
return (_next, p, 0, States.Statement) if (p >= 0) else (None, _LimitLen, 0, States.End)
def _block_comment_begin(s, b):
p = s.find("/*", b)
return p + 1 if (p >= 0) else _LimitLen
def _block_comment_end (s, b):
p = s.find("*/", b)
return (_next, p + len("*/") - 1, len("*/") -1, States.Statement) if (p >= 0) else (None, _LimitLen, 0, States.End)
def _line_comment_begin(s, b):
p = s.find("--", b)
return p + 1 if (p >= 0) else _LimitLen
def _line_comment_end(s, b):
p = s.find("\n", b)
return (_next, p + len("\n") - 1, len("\n") - 1, States.Statement) if (p >= 0) else (None, _LimitLen, 0, States.End)
def _quote_begin(s, b):
next_state = States.Quoted
sq, dq = _singlequote_begin(s, b), _doublequote_begin(s, b)
if (min(sq, dq) == _LimitLen): next_state = States.End
return (_singlequote_end, sq, 0, next_state) if (sq < dq) else (_doublequote_end, dq, 0, next_state)
def _comment_begin(s, b):
bc, lc = _block_comment_begin(s, b), _line_comment_begin(s, b)
if (min(bc, lc) == _LimitLen): next_ = States.End
return (_line_comment_end, lc, 0, States.Comment) if (lc < bc) else (_block_comment_end, bc, 0, States.Comment)
def _next(s, b):
q_func, q_pos, q_ad, q_st = _quote_begin(s, b)
c_func, c_pos, c_ad, c_st = _comment_begin(s, b)
return (q_func, q_pos, 0, q_st) if (q_pos < c_pos) else (c_func, c_pos, 0, c_st)
def strip_comment( st ):
#Évaluation courte
if st == None or len( st.strip() ) == 0: return ""
if ("/*" not in st) and ("--" not in st): return "\n".join( list ( map(lambda x: x.rstrip(), st.split("\n"))))
chars = list( st )
comments = _find_comment_pos( st )
comments.reverse()
for c in comments: del chars[ c[0]: c[1]]
return "\n".join( list( map(lambda x: x.rstrip() , "".join( chars ).split("\n")) ) )
def _find_comment_pos( st ):
cur = -1
begin = 0
comments = []
state = States.Statement
pstate = None
func = _next
while (True):
func, pos, adv, state = func(st, cur + 1)
#Fin de la recherche
if ( state == States.End):
if (pstate == States.Comment):
comments.append((begin, len(st)))
break
cur = pos
#Commencer à commenter->Ignorer le traitement ultérieur
if (state == States.Quoted):
pstate = state
continue
# end comment
if (pstate == States.Comment):
comments.append((begin, pos + adv))
pstate = state
continue
# begin comment
if (state == States.Comment):
begin = pos - 1 # previous a length of single/double quotation
pstate = state
return comments
Je ne me souviens plus de ce que j'ai écrit, alors je vais l'expliquer brièvement.
Strip_comment () et la fonction qui recherche la position du commentaire et retourne une liste de (position de départ, position de fin) _find_comment_pos () Les fonctions principales sont les fonctions et objets alignés sur une ligne avant cela. C'est le helper utilisé par _find_comment_pos ().
Les fonctions diverses étaient à l'origine définies comme des fonctions internes de _find_comment (), mais de cette façon,
func, pos, adv, state = func(st, cur + 1) Lorsque vous exécutez, la première fonction(La fonction suivante à rechercher est renvoyée)Cependant, il semble que ce ne soit pas un objet de fonction normal mais un taple à trois éléments.
Il semble difficile de savoir comment gérer cela, donc je passe docilement à une fonction normale. (Pour cette raison, j'ai décidé de disperser des fonctions supplémentaires dans le module ...)
___find_comment___pos() Variables locales'''func```Contient la fonction à utiliser pour la recherche suivante. Statut actuel et personnages à succès("/**"Et"'"Et改行Et)Cela peut être réalisé en divisant le cas avec, mais comme le branchement conditionnel semble être terrible, j'essaie de renvoyer une fonction de recherche appropriée en fonction du caractère frappé. Exemple) "/"Si vous trouvez, la prochaine chose à rechercher"/"Puisqu'il est décidé, la fonction pour le rechercher_block_comment_end()Est décidé
Est-il suffisant de retourner la position de frappe ensemble à ce moment? J'ai pensé, mais après tout je veux cette information, j'en ai aussi besoin, je dois renvoyer 4 valeurs de chaque fonction...。 Après cela, j'ai ajusté la valeur numérique afin que la position du commentaire puisse être spécifiée par tranche, et l'ai compressée dans la liste. (J'ai oublié les détails..)
Maintenant que la position du commentaire est fixe, il ne vous reste plus qu'à utiliser des tranches pour supprimer le commentaire! !!
La classe string n'a pas de méthode pour supprimer le caractère à la position de la tranche! !!
Ça ne peut pas être aidéchars = list( st )Convertir en liste de caractères avecdel chars[begin:end]Ce faisant, j'ai finalement réussi à supprimer le commentaire, pas la scie à glace tant attendue d.
La dernière instruction return, mais la ligne vide à la fin de la ligne après la suppression du commentaire("Texte; /commentaire/\n"modèle)J'ai décidé d'écrire à nouveau un processus de détournement afin de l'effacer davantage.
#finalement Le fichier utilisé pour le test et le code de test.(sample_connection.txt)Est omis car il a déjà été publié)
sample_multi_statement.sql
select * from SYSLIBL
where TYPE not like = "';%" ;/* comment */
select * from SYSTABLES ; /* "" ' '' ; ;
" */
; -- empty statement
select * from SYSCOLUMNS ;
select * from SYSIBM.SYSDUMMY1;
strip_Pour confirmer l'évaluation de court-circuit du commentaire
sample_multi_statement.sql
select * from SYSLIBL
where TYPE not like = "';%" ;
select * from SYSTABLES ;
;
select * from SYSCOLUMNS ;
select * from SYSIBM.SYSDUMMY1;
sql_helper.py
def test_loading():
from pprint import pprint
file = "sample_multi_statement.sql"
print( f'\ntest load_sql("{file}") ----------------------------')
sts = load_sql( file )
pprint( sts )
file = "sample_multi_statement2.sql"
print( f'\ntest load_sql("{file}") ----------------------------')
sts = load_sql( file )
pprint( sts )
file = "sample_config.txt"
print( f'\ntest load_config("{file}") ----------------------------')
config = load_config( file )
pprint( config )
La quantité de code était plus grande que prévu rien qu'en lecture. J'ai commencé à le créer en supposant qu'il sera utilisé avec IBM i, mais je pense que le contenu jusqu'à présent peut être utilisé comme prétraitement pour d'autres SGBD.
La prochaine fois, nous commencerons du côté de la sortie.
Recommended Posts