Funktionale Programmierung (FP) ist ein Paradigma, bei dem ein Programm aus Funktionen zusammengesetzt ist. Eine Funktion ist ein Baustein, der eine Berechnung kapselt. Eine Funktion, die auf einen Wert angewendet wird, gibt immer denselben berechneten Wert zurück. FP vermeidet die Veränderung des Zustands. FP ermöglicht es Entwicklern, leistungsstarke Verarbeitungspipelines zu erstellen. Die meisten Programmiersprachen unterstützen die funktionale Programmierung, so auch Python. Werfen wir einen Blick darauf, wie FP in Python funktioniert.
Funktionen und Python
Python verwendet das Schlüsselwort lambda, um eine Funktion zu definieren. Wenn wir zum Beispiel f = lambda x: x + 1 eingeben, können wir
eine Funktion namens f definieren, die, wenn sie auf einen Wert angewendet wird, x + 1 zurückgibt:
In [1]: f = lambda x: x + 1
In [2]: f(1)
Out[2]: 2
Funktion Zusammensetzung
Bei der Funktionskomposition geht es um die Kombination von Funktionen. Eine kombinierte Funktion hat die Berechnungseigenschaften beider Funktionen. Wenn wir zum Beispiel zwei Funktionen f und g definieren, können wir sie zu einer Funktion h zusammensetzen, wobei h die Eigenschaften beider Funktionen besitzt.
In [1]: f = lambda x: x + 1
In [2]: g = lambda x: x + 2
In [3]: h = lambda x: f(g(x))
In [4]: h(1)
Out[4]: 4
Reine und unreine Werte
FP beginnt zu glänzen, wenn es mit Werten arbeitet, die eine bestimmte Art von Wert ausdrücken. Die meisten Programme arbeiten mit reinen Werten. Beispiele für reine Werte sind die Zahl 1, der Text hello, oder ein Wert wie True oder False. Mit FP können Sie das Ergebnis einer Berechnung als Wert ausdrücken. Ein solcher Wert ist ein unreiner Wert, weil er eine Auswirkung einer Berechnung ausdrückt. Beispiele für unreine Werte sind Success(123), Failure('First name is empty') oder None.
Über unreine Werte nachdenken
Python bietet Unterstützung für unreine Werte. Ein Beispiel für einen unreinen Wert ist Optional. Eine Methode gibt einen optionalen Wert zurück, um auszudrücken, dass ein Wert nicht immer zurückgegeben wird. Die Methode kann einen Wert 1 oder einen Wert None zurückgeben. Es gibt einen pythonischen Weg, einen solchen Wert zu verarbeiten. Verwenden Sie eine if Anweisung, um den Effekt zu überprüfen und den Wert zu verarbeiten:
In [1]: def return_value_optionally(x: int):
...: if x <= 3:
...: return x
...: else:
...: return None
...:
In [2]: return_value_optionally(1)
Out[2]: 1
In [3]: x = return_value_optionally(5)
In [4]: if x:
...: print(f'pure value: {x}')
...: else:
...: print(f'impure value {x}')
...:
impure value None
Die Art und Weise, wie Python optionale Werte verarbeitet, ist nicht ideal. Python verwendet Anweisungen und mutiert Werte. FP kann helfen, weil Funktionen immer einen Wert zurückgeben und so die Mutation des Zustands vermeiden. Ein weiterer Vorteil ist die Nutzung der Funktionskomposition, die die Verkettung von Funktionen ermöglicht. Durch die Verkettung von Funktionen werden Verarbeitungspipelines erstellt und der kognitive Aufwand für den Entwickler verringert. Verarbeitungspipelines sind einfach zu verstehen.
Funktionen höherer Ordnung
Bevor wir Pipelines erstellen können, müssen wir uns mit Funktionen höherer Ordnung (HoF) befassen. HoF sind Funktionen, die eine Funktion als Argument erhalten oder eine Funktion als Ergebnis zurückgeben. Ein einfaches Konzept, das wir bereits aus der Funktionskomposition kennen:
In [1]: f = lambda x: x + 1
In [2]: g = lambda x: x + 2
In [3]: h = lambda f, g: lambda x: f(g(x))
In [4]: i = h(f, g)
In [5]: i(1)
Out[5]: 4
Funktionale Datenstrukturen
Bevor wir Pipelines erstellen, müssen wir uns mit funktionalen Datenstrukturen (FDS) beschäftigen. FDS unterstützen HoF und ermöglichen es uns, Pipelines zu erstellen. Pipelines ermöglichen es uns, unreine Werte zu verarbeiten. Die Standardbibliothek von Python enthält keine FDS. Ich habe eine kleine Bibliothek erstellt, die die gängigsten FDS-Strukturen enthält. Die Bibliothek heißt python-fp und ist bei Github verfügbar. Die Bibliothek ist im PyPI, dem Python Package Index, verfügbar.
Schauen wir uns an, wie wir einen optionalen Wert auf eine funktionellere Weise verarbeiten können. Wir verwenden fp.option.Option, eine FDS, die die Verarbeitung unreiner Werte als Pipeline unterstützt:
In [1]: from fp.option import Option
In [2]: Option(1)
.map(lambda x: x + 1)
Out[2]: Option(2)
In [3]: Option(1)
.filter(lambda x: x > 1)
Out[3]: Option(None)
In [4]: Option(1)
.bind(lambda x: Option(x + 2))
Out[4]: Option(3)
In [5]: Option.empty()
.map(lambda x: x + 1)
Out[5]: Option(None)
In [6]: Option(1)
.map(lambda x: x + 1)
.fold(0, lambda x: x)
Out[6]: 2
In [7]: Option(1)
.map(lambda x: x + 1)
.filter(lambda x: x > 4)
.fold(0, lambda x: x)
Out[7]: 0
Die Beispiele geben immer einen Wert zurück. Der FDS stellt HoF zur Verfügung, die wir zur Erstellung von Pipelines verwenden. Pipelines sind eine Kette von Operationen, die mit dem Wert, den der FDS enthält, arbeiten. Je nach dem Zustand des FDS wird die Kette von Berechnungen ausgewertet oder nicht.
Um einen unreinen Wert in einen reinen Wert umzuwandeln, müssen wir die Funktion fold verwenden. Fold konvertiert einen unreinen Wert in einen reinen Wert. Je nach FDS-Typ treffen wir eine Entscheidung, was zu tun ist, wenn die FDS leer ist. Welchen reinen Wert möchten wir zum Beispiel zurückgeben, wenn der optionale Wert None ist. Wird es eine Null sein oder ein anderer Wert? Ich habe beschlossen, eine Null zurückzugeben, wenn der Wert None ist, ansonsten gibt die Datenstruktur den reinen Wert zurück, den sie enthält.
Operationen auflisten
Eine Liste ist ebenfalls ein unreiner Wert. Sie drückt den Effekt aus, dass sie null oder mehr Elemente hat. Wir können fp.list.List FDS verwenden, um Pipelines auf der Basis von Listen zu erstellen:
In [1]: from fp.list import List
In [2]: List(1, 2, 3)
.map(lambda x: x + 1)
Out[2]: List(2, 3, 4)
In [3]: List(1, 2, 3)
.filter(lambda x: x > 1)
Out[3]: List(2, 3)
In [4]: List(1, 2, 3)
.bind(lambda x: List(x + 1, x + 2, x + 3))
Out[4]: List(2, 3, 4, 3, 4, 5, 4, 5, 6)
In [5]: List.empty()
.map(lambda x: x + 1)
Out[5]: List()
In [6]: List(1, 2, 3)
.fold(0, lambda c, e: c + e)
Out[6]: 6
In [7]: List(1, 2, 3).sum()
Out[7]: 6
In [8]: List(1, 2, 3)
.intersperse(0)
Out[8]: List(1, 0, 2, 0, 3)
In [9]: List('the', 'book', 'is', 'green')
.mk_string(',')
Out[9]: 'the,book,is,green'
Der Listen-FDS gibt immer einen Wert zurück. Je nachdem, welche Pipeline wir erstellen, gibt die Liste einen unreinen oder einen reinen Wert zurück. Um einen reinen Wert zu erstellen, verwenden Sie map, bind oder head_option.
Eine Liste der Auswirkungen
Wenn wir eine Liste mit unreinen Werten haben, können wir eine Pipeline erstellen, um die unreinen Werte zu kombinieren. Das Ergebnis ist ebenfalls ein unreiner Wert.
In [1]: from fp.list import List
In [2]: from fp.option import Option
In [3]: from fp.sequence import OptionSequence
In [4]: xs = List(Option(1), Option.empty(),
Option(2), Option.empty(), Option(3))
In [5]: OptionSequence.sequence(xs)
Out[5]: Option(List(1, 2, 3))
Nehmen wir an, wir haben eine Liste von Werten, die das Ergebnis einer Validierung darstellen. Die Liste enthält alle unreinen Werte. Einige stehen für leere Werte, andere für reine Werte. Die Operation
Wir können weiter denken und den reinen Wert mit einer Fold-Operation berechnen:
In [6]: result = OptionSequence.sequence(xs)
Out[6]: Option(List(1, 2, 3))
In [7]: ys = result
.fold(List.empty(), lambda x: x)
Out[7]: List(1, 2, 3)
In [8]: ys.fold(0, lambda c, e: c + e)
Out[8]: 6
Werte validieren
Der fp.validation.Validation FDS eignet sich hervorragend zur Überprüfung von Benutzereingaben. Kombiniert mit Option drückt es fehlende Werte aus. Kombiniert mit Regex validiert er Benutzereingaben:
In [1]: from fp.validation import Validation
In [2]: from fp.option import Option
In [3]: import re
In [4]: Validation.from_option(Option.empty(), 'No value')
Out[4]: Failure(err_value='No value', failure=True, success=False)
In [5]: Validation.from_option(Option(1), 'No value')
Out[5]: Success(value=1, failure=False, success=True)
In [6]: Validation.lift(-20, lambda x: x <= 0, 'Number should be positive')
Out[6]: Failure(err_value='Number should be positive', failure=True, success=False)
In [7]: Validation.lift("dennis", lambda x: re.match('[A-Z]*', x), 'name should be all uppercase')
Out[7]: Failure(err_value='name should be all uppercase', failure=True, success=False)
Eine Liste von Überprüfungswerten
Erstellen Sie eine Verarbeitungspipeline zur Verarbeitung einer Liste von Validierungsergebnissen:
In [1]: from fp.list import List
In [2]: from fp.validation import Validation
In [3]: from fp.sequence import ValidationSequence
In [4]: fn = Validation.failure('First name is empty')
In [5]: ln = Validation.failure('Last name is empty')
In [6]: age = Validation.success(27)
In [7]: zipcode = Validation.failure('Invalid zip code')
In [8]: results = ValidationSequence.sequence(List(fn, ln, age, zipcode))
In [9]: results.fold(lambda x: x, lambda x: x)
Out[9]: List(First name is empty, Last name is empty, Invalid zip code)
In [10]: results.fold(lambda x: x, lambda x: x).mk_string(',')
Out[10]: 'First name is empty,Last name is empty,Invalid zip code'
ValidationSequence konstruiert eine Liste von Fehlern und gibt diese zurück, wenn es Validierungsfehler gibt:
In [1]: from fp.list import List
In [2]: from fp.validation import Validation
In [3]: from fp.sequence import ValidationSequence
In [4]: ValidationSequence.sequence(
List(Validation.success('Dennis'),
Validation.success('Vriend'),
Validation.success('43')))
Out[4]: Success(value=List(Dennis, Vriend, 43), failure=False, success=True)
In [5]: result = ValidationSequence.sequence(List(Validation.success('Dennis'), Validation.success('Vriend'), Validation.success('43')))
In [6]: result
.fold(lambda x: x, lambda x: x)
.mk_string(', ')
Out[6]: 'Dennis, Vriend, 43'
Verarbeitung von Boto3-Fehlern Funktionaler Stil
Der Validierungs-FDS fängt Ausnahmen ab, die z.B. von der Boto3-Bibliothek ausgelöst werden. Verwenden Sie das from_try_catch HoF. Boto3 löst Ausnahmen aus, wenn Operationen fehlschlagen. Beim Löschen eines Eimers kann der Vorgang fehlschlagen. Validation fängt den Fehler als Validierungswert ab. Der Listen-FDS fasst alle Fehler zusammen:
In [1]: import boto3
In [2]: from fp.validation import Validation
In [3]: from fp.list import List
In [4]: from fp.sequence import ValidationSequence
In [5]: client = boto3.client('s3')
In [6]: result = List('foo', 'bar', 'baz')
.map(lambda name: Validation
.from_try_catch(lambda: client.delete_bucket(Bucket=name)))
Out[7]: List(
Failure(err_value=ClientError('An error occurred (AccessDenied) when calling the DeleteBucket operation: Access Denied'), failure=True, success=False),
Failure(err_value=ClientError('An error occurred (AccessDenied) when calling the DeleteBucket operation: Access Denied'), failure=True, success=False),
Failure(err_value=ClientError('An error occurred (AccessDenied) when calling the DeleteBucket operation: Access Denied'), failure=True, success=False))
In [8]: errors = ValidationSequence.sequence(result)
Out[8]: Failure(err_value=
List(
An error occurred (AccessDenied) when calling the DeleteBucket operation: Access Denied,
An error occurred (AccessDenied) when calling the DeleteBucket operation: Access Denied,
An error occurred (AccessDenied) when calling the DeleteBucket operation: Access Denied), failure=True, success=False)
In [9]: errors.fold(lambda x: x, lambda x: x).mk_string(',')
Out[9]: 'An error occurred (AccessDenied) when calling the DeleteBucket operation: Access Denied,
An error occurred (AccessDenied) when calling the DeleteBucket operation: Access Denied,
An error occurred (AccessDenied) when calling the DeleteBucket operation: Access Denied'
Refactoring von Pythonic Code
Das folgende Beispiel verwendet den Pythonic-Stil, um eine Sicherheitsgruppe zurückzugeben. Das Prädikat sucht nach tcp-Ports, die von allen Adressen aus Zugriff auf Port 1337 geben:
from typing import *
def inclusive_port_1337(permission: dict) -> bool:
contains_tcp_or_all = permission.get('IpProtocol', '') in ['-1', 'tcp']
can_access_port_1337 = permission.get('FromPort', 65536) <= 1337 <= permission.get('ToPort', -1)
contains_cidr_all = list(filter(lambda r: r['CidrIp'] == '0.0.0.0/0', permission.get('IpRanges', [])))
return contains_tcp_or_all and can_access_port_1337 and contains_cidr_all
def all_ip4_access_to_port_1337(sg: dict) -> Optional[dict]:
return sg if list(filter(lambda p: inclusive_port_1337, sg['IpPermissions'])) else None
Ein FP-Refactor würde folgendermaßen aussehen:
from fp.option import Option
from fp.list import List
def contains_tcp_or_all(permission: dict) -> bool:
return Option(permission.get('IpProtocol'))
.filter(lambda x: x in ['-1', 'tcp'])
.is_defined()
def can_access_port_1337(permission: dict) -> bool:
from_port = Option(permission.get('FromPort')).get_or_else(65536)
to_port = Option(permission.get('ToPort')).get_or_else(-1)
return from_port <= 1337 <= to_port
def contains_cidr_all(permission: dict) -> bool:
return Option(List.from_list(permission.get('IpRanges')))
.get_or_else(List.empty)
.filter(lambda x: x['CidrIp'] == '0.0.0.0/0')
.non_empty()
def filter_tcp_all_access_to_1337(permission: dict) -> bool:
return contains_tcp_or_all(permission)
and can_access_port_1337(permission)
and contains_cidr_all(permission)
def all_ip4_access_to_port_1337(sg: dict) -> List[dict]:
return List.from_list(sg.get('IpPermissions'))
.filter(filter_tcp_all_access_to_1337)
.bind(lambda x: List.from_list(x['IpRanges']).map(lambda x: x['CidrIp']))
Nach dem Refactoring gibt jede Funktion einen Wert zurück. Jede Funktion gibt die Werte, die sie annimmt und zurückgibt, explizit an. Es ist klar, was die Absicht der Pipelines ist. Der Code kann leicht erweitert werden. Jede Funktion kann isoliert getestet werden.
Der folgende Code zeigt eine Pythonic-Methode zur Aggregation von Daten mithilfe einer verschachtelten Liste:
def aggregate_all_tasks_for_cluster_arn() -> List[dict]:
all_tasks = []
for cluster_arn in list_clusters():
for task_arn in list_tasks(cluster_arn):
task = describe_task(task_arn)
if task:
if task['lastStatus'] == 'RUNNING' and task['taskDefinitionArn']:
all_tasks.append(task)
return all_tasks
Der Code verwendet die Liste all_tasks, um die Ergebnisse durch Mutation zu aggregieren. Lassen Sie uns den Code zu FP umstrukturieren:
from fp.list import List
from fp.option import Option
def list_clusters() -> List[str]:
return List('arn:aws:ecs:us-east-1:0000000000:cluster/dev', 'arn:aws:ecs:us-east-1:0000000000:cluster/test')
def list_tasks(cluster_arn: str) -> List[str]:
tasks_per_cluster = {
'arn:aws:ecs:us-east-1:0000000000:cluster/dev': List('arn:aws:ecs:<region>:0000000000:task/c5cba4eb-5dad-405e-96db-71ef8eefe6a8'),
'arn:aws:ecs:us-east-1:0000000000:cluster/test': List('arn:aws:ecs:<region>:0000000000:task/067ef33c-b084-44ad-8217-26512c6c7845')
}
return Option(tasks_per_cluster.get(cluster_arn)).get_or_else(List.empty())
def describe_task(task_arn: str) -> Option[dict]:
task_definition_per_task_arn = {
'arn:aws:ecs:<region>:0000000000:task/c5cba4eb-5dad-405e-96db-71ef8eefe6a8': { 'lastStatus': 'STOPPED' },
'arn:aws:ecs:<region>:0000000000:task/067ef33c-b084-44ad-8217-26512c6c7845': { 'lastStatus': 'RUNNING', 'taskDefinitionArn': 'arn' }
}
return Option(task_definition_per_task_arn.get(task_arn))
def filter_for_running_tasks_with_task_def_arn(task: dict) -> bool:
return Option(task.get('lastStatus')).exists(lambda x: x == 'RUNNING')
and Option(task.get('taskDefinitionArn')).is_defined()
def aggregate_all_tasks_for_cluster_arn() -> List[dict]:
return list_clusters()
.bind(list_tasks)
.bind(lambda x: describe_task(x).fold(List.empty, lambda x: List(x)))
.filter(filter_for_running_tasks_with_task_def_arn)
Nach dem Refactoring ist der Code modularer und drückt seine Absicht klar aus. Alle Funktionen geben immer einen Wert zurück und können mit Hilfe von Pipelines miteinander verkettet werden.
Fazit
Funktionale Programmierung ist ein Paradigma, das mit Python verwendet werden kann. Das Lösen von Problemen auf funktionale Weise führt zu einfachen, aber leistungsstarken Verarbeitungspipelines. Erstellen Sie Pipelines mit funktionalen Datenstrukturen, Funktionen höherer Ordnung und Funktionen. Pipelines sind eine Kette von Funktionen, die immer einen Wert zurückgeben. Pipelines können sowohl reine als auch unreine Werte verarbeiten. Verwenden Sie fold, um eine Pipeline einen reinen Wert zurückgeben zu lassen.
Mit ein paar Zeilen Code können wir leistungsstarke Verarbeitungspipelines ausdrücken, die einfach zu verstehen sind. Ich verwende diese Art der Problemlösung sehr oft. FP spart mir Zeit, reduziert die Komplexität und führt zu einem modularen Code. Ich kann meinen Code jederzeit testen und Funktionen in anderen Umgebungen wiederverwenden. Jetzt können Sie das auch!
Verfasst von
Dennis Vriend
Unsere Ideen
Weitere Blogs
Contact